MDEV-11738: Mariadb uses 100% of several of my 8 cpus doing nothing

MDEV-11581: Mariadb starts InnoDB encryption threads
when key has not changed or data scrubbing turned off

Background: Key rotation is based on background threads
(innodb-encryption-threads) periodically going through
all tablespaces on fil_system. For each tablespace
current used key version is compared to max key age
(innodb-encryption-rotate-key-age). This process
naturally takes CPU. Similarly, in same time need for
scrubbing is investigated. Currently, key rotation
is fully supported on Amazon AWS key management plugin
only but InnoDB does not have knowledge what key
management plugin is used.

This patch re-purposes innodb-encryption-rotate-key-age=0
to disable key rotation and background data scrubbing.
All new tables are added to special list for key rotation
and key rotation is based on sending a event to
background encryption threads instead of using periodic
checking (i.e. timeout).

fil0fil.cc: Added functions fil_space_acquire_low()
to acquire a tablespace when it could be dropped concurrently.
This function is used from fil_space_acquire() or
fil_space_acquire_silent() that will not print
any messages if we try to acquire space that does not exist.
fil_space_release() to release a acquired tablespace.
fil_space_next() to iterate tablespaces in fil_system
using fil_space_acquire() and fil_space_release().
Similarly, fil_space_keyrotation_next() to iterate new
list fil_system->rotation_list where new tables.
are added if key rotation is disabled.
Removed unnecessary functions fil_get_first_space_safe()
fil_get_next_space_safe()

fil_node_open_file(): After page 0 is read read also
crypt_info if it is not yet read.

btr_scrub_lock_dict_func()
buf_page_check_corrupt()
buf_page_encrypt_before_write()
buf_merge_or_delete_for_page()
lock_print_info_all_transactions()
row_fts_psort_info_init()
row_truncate_table_for_mysql()
row_drop_table_for_mysql()
    Use fil_space_acquire()/release() to access fil_space_t.

buf_page_decrypt_after_read():
    Use fil_space_get_crypt_data() because at this point
    we might not yet have read page 0.

fil0crypt.cc/fil0fil.h: Lot of changes. Pass fil_space_t* directly
to functions needing it and store fil_space_t* to rotation state.
Use fil_space_acquire()/release() when iterating tablespaces
and removed unnecessary is_closing from fil_crypt_t. Use
fil_space_t::is_stopping() to detect when access to
tablespace should be stopped. Removed unnecessary
fil_space_get_crypt_data().

fil_space_create(): Inform key rotation that there could
be something to do if key rotation is disabled and new
table with encryption enabled is created.
Remove unnecessary functions fil_get_first_space_safe()
and fil_get_next_space_safe(). fil_space_acquire()
and fil_space_release() are used instead. Moved
fil_space_get_crypt_data() and fil_space_set_crypt_data()
to fil0crypt.cc.

fsp_header_init(): Acquire fil_space_t*, write crypt_data
and release space.

check_table_options()
	Renamed FIL_SPACE_ENCRYPTION_* TO FIL_ENCRYPTION_*

i_s.cc: Added ROTATING_OR_FLUSHING field to
information_schema.innodb_tablespace_encryption
to show current status of key rotation.

23 files changed, 1397 insertions, 1280 deletions

diff --git a/storage/innobase/btr/btr0scrub.cc b/storage/innobase/btr/btr0scrub.cc
index 62a41d19768..560d2ece6c0 100644
--- a/storage/innobase/btr/btr0scrub.cc
+++ b/storage/innobase/btr/btr0scrub.cc
@@ -111,11 +111,18 @@ log_scrub_failure(
 Lock dict mutexes */
 static
 bool
-btr_scrub_lock_dict_func(ulint space, bool lock_to_close_table,
+btr_scrub_lock_dict_func(ulint space_id, bool lock_to_close_table,
 			 const char * file, uint line)
 {
-	uint start = time(0);
-	uint last = start;
+	time_t start = time(0);
+	time_t last = start;
+
+	/* FIXME: this is not the proper way of doing things. The
+	dict_sys->mutex should not be held by any thread for longer
+	than a few microseconds. It must not be held during I/O,
+	for example. So, what is the purpose for this busy-waiting?
+	This function should be rewritten as part of MDEV-8139:
+	Fix scrubbing tests. */
 
 	while (mutex_enter_nowait_func(&(dict_sys->mutex), file, line)) {
 		/* if we lock to close a table, we wait forever
@@ -123,19 +130,24 @@ btr_scrub_lock_dict_func(ulint space, bool lock_to_close_table,
 		* is closing, and then instead give up
 		*/
 		if (lock_to_close_table == false) {
-			if (fil_crypt_is_closing(space)) {
+			fil_space_t* space = fil_space_acquire(space_id);
+			if (!space || space->stop_new_ops) {
+				if (space) {
+					fil_space_release(space);
+				}
 				return false;
 			}
+			fil_space_release(space);
 		}
 		os_thread_sleep(250000);
 
-		uint now = time(0);
+		time_t now = time(0);
 		if (now >= last + 30) {
 			fprintf(stderr,
-				"WARNING: %s:%u waited %u seconds for"
+				"WARNING: %s:%u waited %ld seconds for"
 				" dict_sys lock, space: %lu"
-				" lock_to_close_table: %u\n",
-				file, line, now - start, space,
+				" lock_to_close_table: %d\n",
+				file, line, now - start, space_id,
 				lock_to_close_table);
 
 			last = now;
@@ -181,16 +193,24 @@ void
 btr_scrub_table_close_for_thread(
 	btr_scrub_t *scrub_data)
 {
-	if (scrub_data->current_table == NULL)
+	if (scrub_data->current_table == NULL) {
 		return;
+	}
 
-	bool lock_for_close = true;
-	btr_scrub_lock_dict(scrub_data->space, lock_for_close);
+	fil_space_t* space = fil_space_acquire(scrub_data->space);
 
-	/* perform the actual closing */
-	btr_scrub_table_close(scrub_data->current_table);
+	/* If tablespace is not marked as stopping perform
+	the actual close. */
+	if (space && !space->is_stopping()) {
+		mutex_enter(&dict_sys->mutex);
+		/* perform the actual closing */
+		btr_scrub_table_close(scrub_data->current_table);
+		mutex_exit(&dict_sys->mutex);
+	}
 
-	btr_scrub_unlock_dict();
+	if (space) {
+		fil_space_release(space);
+	}
 
 	scrub_data->current_table = NULL;
 	scrub_data->current_index = NULL;
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 9e07a4d70f0..c7e2183fa7b 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -626,7 +626,7 @@ buf_page_is_corrupted(
 	ulint		checksum_field2;
 	ulint 		space_id = mach_read_from_4(
 		read_buf + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
-	ulint page_type = mach_read_from_4(
+	ulint page_type = mach_read_from_2(
 		read_buf + FIL_PAGE_TYPE);
 
 	/* We can trust page type if page compression is set on tablespace
@@ -4480,11 +4480,14 @@ buf_page_check_corrupt(
 	byte* dst_frame = (zip_size) ? bpage->zip.data :
 		((buf_block_t*) bpage)->frame;
 	ulint space_id = bpage->space;
-	fil_space_t* space = fil_space_found_by_id(space_id);
-	fil_space_crypt_t* crypt_data = space->crypt_data;
+	fil_space_t* space = fil_space_acquire_silent(space_id);
 	bool still_encrypted = false;
 	bool corrupted = false;
 	ulint page_type = mach_read_from_2(dst_frame + FIL_PAGE_TYPE);
+	fil_space_crypt_t* crypt_data = NULL;
+
+	ut_ad(space);
+	crypt_data = space->crypt_data;
 
 	/* In buf_decrypt_after_read we have either decrypted the page if
 	page post encryption checksum matches and used key_id is found
@@ -4513,7 +4516,7 @@ buf_page_check_corrupt(
 		ib_logf(IB_LOG_LEVEL_ERROR,
 			"%s: Block in space_id " ULINTPF " in file %s corrupted.",
 			page_type ==  FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED ? "Maybe corruption" : "Corruption",
-			space_id, space->name ? space->name : "NULL");
+			space_id, (space && space->name) ? space->name : "NULL");
 		ib_logf(IB_LOG_LEVEL_ERROR,
 			"Based on page type %s (" ULINTPF ")",
 			fil_get_page_type_name(page_type), page_type);
@@ -4523,7 +4526,7 @@ buf_page_check_corrupt(
 
 		ib_logf(IB_LOG_LEVEL_ERROR,
 			"Block in space_id " ULINTPF " in file %s encrypted.",
-			space_id, space->name ? space->name : "NULL");
+			space_id, (space && space->name) ? space->name : "NULL");
 		ib_logf(IB_LOG_LEVEL_ERROR,
 				"However key management plugin or used key_version %u is not found or"
 				" used encryption algorithm or method does not match.",
@@ -4535,6 +4538,10 @@ buf_page_check_corrupt(
 		}
 	}
 
+	if (space) {
+		fil_space_release(space);
+	}
+
 	return corrupted;
 }
 
@@ -6088,13 +6095,6 @@ buf_page_encrypt_before_write(
 	byte*		src_frame,
 	ulint		space_id)
 {
-	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space_id);
-	ulint zip_size = buf_page_get_zip_size(bpage);
-	ulint page_size = (zip_size) ? zip_size : UNIV_PAGE_SIZE;
-	buf_pool_t* buf_pool = buf_pool_from_bpage(bpage);
-	bool page_compressed = fil_space_is_page_compressed(bpage->space);
-	bool encrypted = true;
-
 	bpage->real_size = UNIV_PAGE_SIZE;
 
 	fil_page_type_validate(src_frame);
@@ -6111,7 +6111,20 @@ buf_page_encrypt_before_write(
 		return src_frame;
 	}
 
-	if (crypt_data != NULL && crypt_data->not_encrypted()) {
+	fil_space_t* space = fil_space_acquire_silent(space_id);
+
+	/* Tablespace must exist during write operation */
+	if (!space) {
+		/* This could be true on discard if we have injected a error
+		case e.g. in innodb.innodb-wl5522-debug-zip so that space
+		is already marked as stop_new_ops = true. */
+		return src_frame;
+	}
+
+	fil_space_crypt_t* crypt_data = space->crypt_data;
+	bool encrypted = true;
+
+	if (space->crypt_data != NULL && space->crypt_data->not_encrypted()) {
 		/* Encryption is disabled */
 		encrypted = false;
 	}
@@ -6128,11 +6141,17 @@ buf_page_encrypt_before_write(
 		encrypted = false;
 	}
 
+	bool page_compressed = fil_space_is_page_compressed(bpage->space);
+
 	if (!encrypted && !page_compressed) {
 		/* No need to encrypt or page compress the page */
+		fil_space_release(space);
 		return src_frame;
 	}
 
+	ulint zip_size = buf_page_get_zip_size(bpage);
+	ulint page_size = (zip_size) ? zip_size : UNIV_PAGE_SIZE;
+	buf_pool_t* buf_pool = buf_pool_from_bpage(bpage);
 	/* Find free slot from temporary memory array */
 	buf_tmp_buffer_t* slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
 	slot->out_buf = NULL;
@@ -6142,11 +6161,10 @@ buf_page_encrypt_before_write(
 
 	if (!page_compressed) {
 		/* Encrypt page content */
-		byte* tmp = fil_space_encrypt(bpage->space,
+		byte* tmp = fil_space_encrypt(space,
 					      bpage->offset,
 					      bpage->newest_modification,
 					      src_frame,
-					      zip_size,
 					      dst_frame);
 
 		ulint key_version = mach_read_from_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
@@ -6184,11 +6202,10 @@ buf_page_encrypt_before_write(
 		if(encrypted) {
 
 			/* And then we encrypt the page content */
-			tmp = fil_space_encrypt(bpage->space,
+			tmp = fil_space_encrypt(space,
 						bpage->offset,
 						bpage->newest_modification,
 						tmp,
-						zip_size,
 						dst_frame);
 		}
 
@@ -6199,6 +6216,7 @@ buf_page_encrypt_before_write(
 	fil_page_type_validate(dst_frame);
 #endif
 
+	fil_space_release(space);
 	// return dst_frame which will be written
 	return dst_frame;
 }
@@ -6224,10 +6242,16 @@ buf_page_decrypt_after_read(
 	bool page_compressed_encrypted = fil_page_is_compressed_encrypted(dst_frame);
 	buf_pool_t* buf_pool = buf_pool_from_bpage(bpage);
 	bool success = true;
-	ulint 		space_id = mach_read_from_4(
-		dst_frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
-	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space_id);
 
+	bpage->key_version = key_version;
+
+	if (bpage->offset == 0) {
+		/* File header pages are not encrypted/compressed */
+		return (true);
+	}
+
+	fil_space_t* space = fil_space_acquire(bpage->space);
+	fil_space_crypt_t* crypt_data = space->crypt_data;
 
 	/* Page is encrypted if encryption information is found from
 	tablespace and page contains used key_version. This is true
@@ -6236,13 +6260,6 @@ buf_page_decrypt_after_read(
 		key_version = 0;
 	}
 
-	bpage->key_version = key_version;
-
-	if (bpage->offset == 0) {
-		/* File header pages are not encrypted/compressed */
-		return (true);
-	}
-
 	if (page_compressed) {
 		/* the page we read is unencrypted */
 		/* Find free slot from temporary memory array */
@@ -6284,15 +6301,15 @@ buf_page_decrypt_after_read(
 #endif
 
 			/* decrypt using crypt_buf to dst_frame */
-			byte* res = fil_space_decrypt(bpage->space,
+			byte* res = fil_space_decrypt(space,
 						slot->crypt_buf,
-						size,
 						dst_frame,
 						&bpage->encrypted);
 
 			if (!res) {
 				success = false;
 			}
+
 #ifdef UNIV_DEBUG
 			fil_page_type_validate(dst_frame);
 #endif
@@ -6323,5 +6340,6 @@ buf_page_decrypt_after_read(
 		}
 	}
 
+	fil_space_release(space);
 	return (success);
 }
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 8a4c0bec1e1..7bef91f47ff 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -37,7 +37,6 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 #include "fsp0fsp.h"
 #include "fil0pagecompress.h"
 #include "ha_prototypes.h" // IB_LOG_
-
 #include <my_crypt.h>
 
 /** Mutex for keys */
@@ -59,7 +58,7 @@ UNIV_INTERN uint srv_n_fil_crypt_threads = 0;
 UNIV_INTERN uint srv_n_fil_crypt_threads_started = 0;
 
 /** At this age or older a space/page will be rotated */
-UNIV_INTERN uint srv_fil_crypt_rotate_key_age = 1;
+UNIV_INTERN uint srv_fil_crypt_rotate_key_age;
 
 /** Event to signal FROM the key rotation threads. */
 static os_event_t fil_crypt_event;
@@ -67,11 +66,11 @@ static os_event_t fil_crypt_event;
 /** Event to signal TO the key rotation threads. */
 UNIV_INTERN os_event_t fil_crypt_threads_event;
 
-/** Event for waking up threads throttle */
+/** Event for waking up threads throttle. */
 static os_event_t fil_crypt_throttle_sleep_event;
 
-/** Mutex for key rotation threads */
-static ib_mutex_t fil_crypt_threads_mutex;
+/** Mutex for key rotation threads. */
+UNIV_INTERN ib_mutex_t fil_crypt_threads_mutex;
 
 #ifdef UNIV_PFS_MUTEX
 static mysql_pfs_key_t fil_crypt_threads_mutex_key;
@@ -104,9 +103,12 @@ static mysql_pfs_key_t fil_crypt_stat_mutex_key;
 UNIV_INTERN mysql_pfs_key_t fil_crypt_data_mutex_key;
 #endif
 
+/** Is background scrubbing enabled, defined on btr0scrub.cc */
+extern my_bool srv_background_scrub_data_uncompressed;
+extern my_bool srv_background_scrub_data_compressed;
+
 static bool
 fil_crypt_needs_rotation(
-/*=====================*/
 	fil_encryption_t        encrypt_mode,           /*!< in: Encryption
 							mode */
 	uint			key_version,		/*!< in: Key version */
@@ -118,7 +120,6 @@ Init space crypt */
 UNIV_INTERN
 void
 fil_space_crypt_init()
-/*==================*/
 {
 	mutex_create(fil_crypt_key_mutex_key,
 		     &fil_crypt_key_mutex, SYNC_NO_ORDER_CHECK);
@@ -127,6 +128,7 @@ fil_space_crypt_init()
 
 	mutex_create(fil_crypt_stat_mutex_key,
 		     &crypt_stat_mutex, SYNC_NO_ORDER_CHECK);
+
 	memset(&crypt_stat, 0, sizeof(crypt_stat));
 }
 
@@ -135,7 +137,6 @@ Cleanup space crypt */
 UNIV_INTERN
 void
 fil_space_crypt_cleanup()
-/*=====================*/
 {
 	os_event_free(fil_crypt_throttle_sleep_event);
 	fil_crypt_throttle_sleep_event = NULL;
@@ -147,7 +148,7 @@ fil_space_crypt_cleanup()
 Get latest key version from encryption plugin.
 @return key version or ENCRYPTION_KEY_VERSION_INVALID */
 uint
-fil_space_crypt_struct::key_get_latest_version(void)
+fil_space_crypt_t::key_get_latest_version(void)
 {
 	uint key_version = key_found;
 
@@ -161,12 +162,12 @@ fil_space_crypt_struct::key_get_latest_version(void)
 }
 
 /******************************************************************
-Get the latest(key-version), waking the encrypt thread, if needed */
+Get the latest(key-version), waking the encrypt thread, if needed
+@param[in,out]	crypt_data	Crypt data */
 static inline
 uint
 fil_crypt_get_latest_key_version(
-/*=============================*/
-	fil_space_crypt_t* crypt_data) 	/*!< in: crypt data */
+	fil_space_crypt_t* crypt_data)
 {
 	ut_ad(crypt_data != NULL);
 
@@ -205,28 +206,31 @@ crypt_data_scheme_locker(
 
 /******************************************************************
 Create a fil_space_crypt_t object
+@param[in]	type		CRYPT_SCHEME_UNENCRYPTE or
+				CRYPT_SCHEME_1
+@param[in]	encrypt_mode	FIL_ENCRYPTION_DEFAULT or
+				FIL_ENCRYPTION_ON or
+				FIL_ENCRYPTION_OFF
+@param[in]	min_key_version key_version or 0
+@param[in]	key_id		Used key id
 @return crypt object */
 static
 fil_space_crypt_t*
 fil_space_create_crypt_data(
-/*========================*/
 	uint			type,
 	fil_encryption_t	encrypt_mode,
 	uint			min_key_version,
-	uint			key_id,
-	ulint			offset)
+	uint			key_id)
 {
-	const uint sz = sizeof(fil_space_crypt_t);
-	void* buf = mem_zalloc(sz);
+	void* buf = mem_zalloc(sizeof(fil_space_crypt_t));
 	fil_space_crypt_t* crypt_data = NULL;
 
 	if (buf) {
 		crypt_data = new(buf)
-			fil_space_crypt_struct(
+			fil_space_crypt_t(
 				type,
 				min_key_version,
 				key_id,
-				offset,
 				encrypt_mode);
 	}
 
@@ -235,25 +239,30 @@ fil_space_create_crypt_data(
 
 /******************************************************************
 Create a fil_space_crypt_t object
+@param[in]	encrypt_mode	FIL_ENCRYPTION_DEFAULT or
+				FIL_ENCRYPTION_ON or
+				FIL_ENCRYPTION_OFF
+
+@param[in]	key_id		Encryption key id
 @return crypt object */
 UNIV_INTERN
 fil_space_crypt_t*
 fil_space_create_crypt_data(
-/*========================*/
-	fil_encryption_t	encrypt_mode,	/*!< in: encryption mode */
-	uint			key_id)		/*!< in: encryption key id */
+	fil_encryption_t	encrypt_mode,
+	uint			key_id)
 {
-	return (fil_space_create_crypt_data(0, encrypt_mode, 0, key_id, 0));
+	return (fil_space_create_crypt_data(0, encrypt_mode, 0, key_id));
 }
 
 /******************************************************************
-Merge fil_space_crypt_t object */
+Merge fil_space_crypt_t object
+@param[in,out]	dst		Destination cryp data
+@param[in]	src		Source crypt data */
 UNIV_INTERN
 void
 fil_space_merge_crypt_data(
-/*=======================*/
-	fil_space_crypt_t* dst,/*!< out: Crypt data */
-	const fil_space_crypt_t* src)/*!< in: Crypt data */
+	fil_space_crypt_t* dst,
+	const fil_space_crypt_t* src)
 {
 	mutex_enter(&dst->mutex);
 
@@ -268,21 +277,22 @@ fil_space_merge_crypt_data(
 	dst->type = src->type;
 	dst->min_key_version = src->min_key_version;
 	dst->keyserver_requests += src->keyserver_requests;
-	dst->closing = src->closing;
 
 	mutex_exit(&dst->mutex);
 }
 
 /******************************************************************
 Read crypt data from a page (0)
-@return crypt data from page 0. */
+@param[in]	space		space_id
+@param[in]	page		Page 0
+@param[in]	offset		Offset to crypt data
+@return crypt data from page 0 or NULL. */
 UNIV_INTERN
 fil_space_crypt_t*
 fil_space_read_crypt_data(
-/*======================*/
-	ulint		space,	/*!< in: file space id*/
-	const byte*	page,	/*!< in: page 0 */
-	ulint		offset)	/*!< in: offset */
+	ulint		space,
+	const byte*	page,
+	ulint		offset)
 {
 	if (memcmp(page + offset, CRYPT_MAGIC, MAGIC_SZ) != 0) {
 		/* Crypt data is not stored. */
@@ -295,8 +305,8 @@ fil_space_read_crypt_data(
 	       type == CRYPT_SCHEME_1)) {
 
 		ib_logf(IB_LOG_LEVEL_ERROR,
-			"Found non sensible crypt scheme: %lu for space %lu "
-			" offset: %lu bytes: "
+			"Found non sensible crypt scheme: " ULINTPF " for space " ULINTPF
+			" offset: " ULINTPF " bytes: "
 			"[ %.2x %.2x %.2x %.2x %.2x %.2x ].",
 			type, space, offset,
 			page[offset + 0 + MAGIC_SZ],
@@ -347,43 +357,37 @@ fil_space_read_crypt_data(
 }
 
 /******************************************************************
-Free a crypt data object */
+Free a crypt data object
+@param[in,out] crypt_data	crypt data to be freed */
 UNIV_INTERN
 void
 fil_space_destroy_crypt_data(
-/*=========================*/
-	fil_space_crypt_t **crypt_data)	/*!< out: crypt data */
+	fil_space_crypt_t **crypt_data)
 {
 	if (crypt_data != NULL && (*crypt_data) != NULL) {
 		fil_space_crypt_t* c = *crypt_data;
-		c->~fil_space_crypt_struct();
+		c->~fil_space_crypt_t();
 		mem_free(c);
 		*crypt_data = NULL;
 	}
 }
 
 /******************************************************************
-Write crypt data to a page (0) */
-static
+Write crypt data to a page (0)
+@param[in,out]	page0		Page 0 where to write
+@param[in,out]	mtr		Minitransaction */
+UNIV_INTERN
 void
-fil_space_write_crypt_data_low(
-/*===========================*/
-	fil_space_crypt_t*	crypt_data,	/*<! out: crypt data */
-	ulint			type,		/*<! in: crypt scheme */
-	byte* 			page,		/*<! in: page 0 */
-	ulint			offset,		/*<! in: offset */
-	ulint			maxsize,	/*<! in: size of crypt data */
-	mtr_t*			mtr)		/*<! in: minitransaction */
+fil_space_crypt_t::write_page0(
+	byte* 			page,
+	mtr_t*			mtr)
 {
-	ut_a(offset > 0 && offset < UNIV_PAGE_SIZE);
 	ulint space_id = mach_read_from_4(
 		page + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
-	const uint len = sizeof(crypt_data->iv);
-	const uint min_key_version = crypt_data->min_key_version;
-	const uint key_id = crypt_data->key_id;
-	const fil_encryption_t encryption = crypt_data->encryption;
-	crypt_data->page0_offset = offset;
-	ut_a(2 + len + 4 + 1 + 4 + MAGIC_SZ < maxsize);
+	const uint len = sizeof(iv);
+	ulint zip_size = fsp_header_get_zip_size(page);
+	const ulint offset = fsp_header_get_crypt_offset(zip_size);
+	page0_offset = offset;
 
 	/*
 	redo log this as bytewise updates to page 0
@@ -393,7 +397,7 @@ fil_space_write_crypt_data_low(
 	mlog_write_string(page + offset, CRYPT_MAGIC, MAGIC_SZ, mtr);
 	mlog_write_ulint(page + offset + MAGIC_SZ + 0, type, MLOG_1BYTE, mtr);
 	mlog_write_ulint(page + offset + MAGIC_SZ + 1, len, MLOG_1BYTE, mtr);
-	mlog_write_string(page + offset + MAGIC_SZ + 2, crypt_data->iv, len,
+	mlog_write_string(page + offset + MAGIC_SZ + 2, iv, len,
 			  mtr);
 	mlog_write_ulint(page + offset + MAGIC_SZ + 2 + len, min_key_version,
 			 MLOG_4BYTES, mtr);
@@ -425,44 +429,61 @@ fil_space_write_crypt_data_low(
 		log_ptr += 1;
 		mlog_close(mtr, log_ptr);
 
-		mlog_catenate_string(mtr, crypt_data->iv, len);
+		mlog_catenate_string(mtr, iv, len);
 	}
 }
 
 /******************************************************************
-Write crypt data to a page (0) */
-UNIV_INTERN
-void
-fil_space_write_crypt_data(
-/*=======================*/
-	ulint			space,		/*<! in: file space */
-	byte* 			page,		/*<! in: page 0 */
-	ulint			offset,		/*<! in: offset */
-	ulint			maxsize,	/*<! in: size of crypt data */
-	mtr_t*			mtr)		/*<! in: minitransaction */
+Set crypt data for a tablespace
+@param[in,out]		space		Tablespace
+@param[in,out]		crypt_data	Crypt data to be set
+@return crypt_data in tablespace */
+static
+fil_space_crypt_t*
+fil_space_set_crypt_data(
+	fil_space_t*		space,
+	fil_space_crypt_t*	crypt_data)
 {
-	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space);
+	fil_space_crypt_t* free_crypt_data = NULL;
+	fil_space_crypt_t* ret_crypt_data = NULL;
+
+	/* Provided space is protected using fil_space_acquire()
+	from concurrent operations. */
+	if (space->crypt_data != NULL) {
+		/* There is already crypt data present,
+		merge new crypt_data */
+		fil_space_merge_crypt_data(space->crypt_data,
+						   crypt_data);
+		ret_crypt_data = space->crypt_data;
+		free_crypt_data = crypt_data;
+	} else {
+		space->crypt_data = crypt_data;
+		ret_crypt_data = space->crypt_data;
+	}
 
-	/* If no crypt data is stored on memory cache for this space,
-	then do not continue writing crypt data to page 0. */
-	if (crypt_data == NULL) {
-		return;
+	if (free_crypt_data != NULL) {
+		/* there was already crypt data present and the new crypt
+		* data provided as argument to this function has been merged
+		* into that => free new crypt data
+		*/
+		fil_space_destroy_crypt_data(&free_crypt_data);
 	}
 
-	fil_space_write_crypt_data_low(crypt_data, crypt_data->type,
-				       page, offset, maxsize, mtr);
+	return ret_crypt_data;
 }
 
 /******************************************************************
 Parse a MLOG_FILE_WRITE_CRYPT_DATA log entry
+@param[in]	ptr		Log entry start
+@param[in]	end_ptr		Log entry end
+@param[in]	block		buffer block
 @return position on log buffer */
 UNIV_INTERN
-byte*
+const byte*
 fil_parse_write_crypt_data(
-/*=======================*/
-	byte*		ptr,	/*!< in: Log entry start */
-	byte*		end_ptr,/*!< in: Log entry end */
-	buf_block_t*	block)	/*!< in: buffer block */
+	const byte*		ptr,
+	const byte*		end_ptr,
+	const buf_block_t*	block)
 {
 	/* check that redo log entry is complete */
 	uint entry_size =
@@ -513,47 +534,36 @@ fil_parse_write_crypt_data(
 	ptr += len;
 
 	/* update fil_space memory cache with crypt_data */
-	fil_space_set_crypt_data(space_id, crypt_data);
+	fil_space_t* space = fil_space_acquire_silent(space_id);
 
-	return ptr;
-}
+	if (space) {
+		crypt_data = fil_space_set_crypt_data(space, crypt_data);
+		fil_space_release(space);
+	}
 
-/******************************************************************
-Clear crypt data from a page (0) */
-UNIV_INTERN
-void
-fil_space_clear_crypt_data(
-/*=======================*/
-	byte*	page, 	/*!< in/out: Page 0 */
-	ulint 	offset)	/*!< in: Offset */
-{
-	//TODO(jonaso): pass crypt-data and read len from there
-	ulint len = CRYPT_SCHEME_1_IV_LEN;
-	ulint size =
-		sizeof(CRYPT_MAGIC) +
-		1 +   // type
-		1 +   // len
-		len + // iv
-		4 +    // min key version
-		4 +    // key id
-		1; // fil_encryption_t
-	memset(page + offset, 0, size);
+	return ptr;
 }
 
 /******************************************************************
-Encrypt a buffer */
+Encrypt a buffer
+@param[in,out]		crypt_data	Crypt data
+@param[in]		space		space_id
+@param[in]		offset		Page offset
+@param[in]		lsn		Log sequence number
+@param[in]		src_frame	Page to encrypt
+@param[in]		zip_size	Compressed size or 0
+@param[in,out]		dst_frame	Output buffer
+@return encrypted buffer or NULL */
 UNIV_INTERN
 byte*
 fil_encrypt_buf(
-/*============*/
-	fil_space_crypt_t* crypt_data,	/*!< in: crypt data */
-	ulint		space,		/*!< in: Space id */
-	ulint		offset,		/*!< in: Page offset */
-	lsn_t		lsn,		/*!< in: lsn */
-	byte*		src_frame,	/*!< in: Source page to be encrypted */
-	ulint		zip_size,	/*!< in: compressed size if
-					row format compressed */
-	byte*		dst_frame)	/*!< in: outbut buffer */
+	fil_space_crypt_t* crypt_data,
+	ulint		space,
+	ulint		offset,
+	lsn_t		lsn,
+	const byte*	src_frame,
+	ulint		zip_size,
+	byte*		dst_frame)
 {
 	ulint page_size = (zip_size) ? zip_size : UNIV_PAGE_SIZE;
 	uint key_version = fil_crypt_get_latest_key_version(crypt_data);
@@ -634,40 +644,40 @@ fil_encrypt_buf(
 }
 
 /******************************************************************
-Encrypt a page */
+Encrypt a page
+
+@param[in]		space		Tablespace
+@param[in]		offset		Page offset
+@param[in]		lsn		Log sequence number
+@param[in]		src_frame	Page to encrypt
+@param[in,out]		dst_frame	Output buffer
+@return encrypted buffer or NULL */
 UNIV_INTERN
 byte*
 fil_space_encrypt(
-/*==============*/
-	ulint		space,		/*!< in: Space id */
-	ulint		offset,		/*!< in: Page offset */
-	lsn_t		lsn,		/*!< in: lsn */
-	byte*		src_frame,	/*!< in: Source page to be encrypted */
-	ulint		zip_size,	/*!< in: compressed size if
-					row_format compressed */
-	byte*		dst_frame)	/*!< in: outbut buffer */
+	const fil_space_t*	space,
+	ulint		offset,
+	lsn_t		lsn,
+	byte*		src_frame,
+	byte*		dst_frame)
 {
-	fil_space_crypt_t* crypt_data = NULL;
-
 	ulint orig_page_type = mach_read_from_2(src_frame+FIL_PAGE_TYPE);
 
 	if (orig_page_type==FIL_PAGE_TYPE_FSP_HDR
 		|| orig_page_type==FIL_PAGE_TYPE_XDES) {
 		/* File space header or extent descriptor do not need to be
 		encrypted. */
-		return src_frame;
+		return (src_frame);
 	}
 
-	/* Get crypt data from file space */
-	crypt_data = fil_space_get_crypt_data(space);
-
-	if (crypt_data == NULL) {
-		return src_frame;
+	if (!space->crypt_data || !space->crypt_data->is_encrypted()) {
+		return (src_frame);
 	}
 
-	ut_a(crypt_data != NULL && crypt_data->is_encrypted());
-
-	byte* tmp = fil_encrypt_buf(crypt_data, space, offset, lsn, src_frame, zip_size, dst_frame);
+	fil_space_crypt_t* crypt_data = space->crypt_data;
+	ut_ad(space->n_pending_ops);
+	ulint zip_size = fsp_flags_get_zip_size(space->flags);
+	byte* tmp = fil_encrypt_buf(crypt_data, space->id, offset, lsn, src_frame, zip_size, dst_frame);
 
 #ifdef UNIV_DEBUG
 	if (tmp) {
@@ -679,7 +689,6 @@ fil_space_encrypt(
 		byte* comp_mem = NULL;
 		byte* uncomp_mem = NULL;
 		ulint size = (zip_size) ? zip_size : UNIV_PAGE_SIZE;
-		fil_space_t* tspace = fil_space_found_by_id(space);
 
 		if (page_compressed_encrypted) {
 			comp_mem = (byte *)malloc(UNIV_PAGE_SIZE);
@@ -689,7 +698,7 @@ fil_space_encrypt(
 			src = uncomp_mem;
 		}
 
-		bool corrupted1 = buf_page_is_corrupted(true, src, zip_size, tspace);
+		bool corrupted1 = buf_page_is_corrupted(true, src, zip_size, space);
 		bool ok = fil_space_decrypt(crypt_data, tmp_mem, size, tmp, &err);
 
 		/* Need to decompress the page if it was also compressed */
@@ -698,18 +707,17 @@ fil_space_encrypt(
 			fil_decompress_page(tmp_mem, comp_mem, UNIV_PAGE_SIZE, NULL);
 		}
 
-		bool corrupted = buf_page_is_corrupted(true, tmp_mem, zip_size, tspace);
+		bool corrupted = buf_page_is_corrupted(true, tmp_mem, zip_size, space);
 		bool different = memcmp(src, tmp_mem, size);
 
 		if (!ok || corrupted || corrupted1 || err != DB_SUCCESS || different) {
-			fprintf(stderr, "JAN: ok %d corrupted %d corrupted1 %d err %d different %d\n", ok , corrupted, corrupted1, err, different);
-			fprintf(stderr, "JAN1: src_frame\n");
+			fprintf(stderr, "ok %d corrupted %d corrupted1 %d err %d different %d\n", ok , corrupted, corrupted1, err, different);
+			fprintf(stderr, "src_frame\n");
 			buf_page_print(src_frame, zip_size, BUF_PAGE_PRINT_NO_CRASH);
-			fprintf(stderr, "JAN2: encrypted_frame\n");
+			fprintf(stderr, "encrypted_frame\n");
 			buf_page_print(tmp, zip_size, BUF_PAGE_PRINT_NO_CRASH);
-			fprintf(stderr, "JAN1: decrypted_frame\n");
-			buf_page_print(tmp_mem, zip_size, BUF_PAGE_PRINT_NO_CRASH);
-			ut_error;
+			fprintf(stderr, "decrypted_frame\n");
+			buf_page_print(tmp_mem, zip_size, 0);
 		}
 
 		free(tmp_mem);
@@ -728,45 +736,22 @@ fil_space_encrypt(
 	return tmp;
 }
 
-/*********************************************************************
-Check if extra buffer shall be allocated for decrypting after read
-@return true if fil space has encryption data. */
-UNIV_INTERN
-bool
-fil_space_check_encryption_read(
-/*=============================*/
-	ulint space)          /*!< in: tablespace id */
-{
-	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space);
-
-	if (crypt_data == NULL) {
-		return false;
-	}
-
-	if (crypt_data->type == CRYPT_SCHEME_UNENCRYPTED) {
-		return false;
-	}
-
-	if (crypt_data->not_encrypted()) {
-		return false;
-	}
-
-	return true;
-}
-
 /******************************************************************
 Decrypt a page
+@param[in]	crypt_data		crypt_data
+@param[in]	tmp_frame		Temporary buffer
+@param[in]	page_size		Page size
+@param[in,out]	src_frame		Page to decrypt
+@param[out]	err			DB_SUCCESS or DB_DECRYPTION_FAILED
 @return true if page decrypted, false if not.*/
 UNIV_INTERN
 bool
 fil_space_decrypt(
-/*==============*/
-	fil_space_crypt_t*	crypt_data,	/*!< in: crypt data */
-	byte*			tmp_frame,	/*!< in: temporary buffer */
-	ulint			page_size,	/*!< in: page size */
-	byte*			src_frame,	/*!< in: out: page buffer */
-	dberr_t*		err)		/*!< in: out: DB_SUCCESS or
-						error code */
+	fil_space_crypt_t*	crypt_data,
+	byte*			tmp_frame,
+	ulint			page_size,
+	byte*			src_frame,
+	dberr_t*		err)
 {
 	ulint page_type = mach_read_from_2(src_frame+FIL_PAGE_TYPE);
 	uint key_version = mach_read_from_4(src_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
@@ -774,6 +759,7 @@ fil_space_decrypt(
 	ulint offset = mach_read_from_4(src_frame + FIL_PAGE_OFFSET);
 	ulint space = mach_read_from_4(src_frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
 	ib_uint64_t lsn = mach_read_from_8(src_frame + FIL_PAGE_LSN);
+
 	*err = DB_SUCCESS;
 
 	if (key_version == ENCRYPTION_KEY_NOT_ENCRYPTED) {
@@ -788,12 +774,12 @@ fil_space_decrypt(
 			first page in a system tablespace
 			data file (ibdata*, not *.ibd), if not
 			clear it. */
-#ifdef UNIV_DEBUG
-			ib_logf(IB_LOG_LEVEL_WARN,
-				"Page on space %lu offset %lu has key_version %u"
+
+			DBUG_PRINT("ib_crypt",
+				("Page on space %lu offset %lu has key_version %u"
 				" when it shoud be undefined.",
-				space, offset, key_version);
-#endif
+				space, offset, key_version));
+
 			mach_write_to_4(src_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 0);
 		}
 		return false;
@@ -862,7 +848,7 @@ fil_space_decrypt(
 
 /******************************************************************
 Decrypt a page
-@param[in]	space			Tablespace id
+@param[in]	space			Tablespace
 @param[in]	tmp_frame		Temporary buffer used for decrypting
 @param[in]	page_size		Page size
 @param[in,out]	src_frame		Page to decrypt
@@ -872,20 +858,24 @@ not needed.*/
 UNIV_INTERN
 byte*
 fil_space_decrypt(
-	ulint		space,
+	const fil_space_t* space,
 	byte*		tmp_frame,
-	ulint		page_size,
 	byte*		src_frame,
 	bool*		decrypted)
 {
 	dberr_t err = DB_SUCCESS;
 	byte* res = NULL;
+	ulint zip_size = fsp_flags_get_zip_size(space->flags);
+	ulint size = zip_size ? zip_size : UNIV_PAGE_SIZE;
 	*decrypted = false;
 
+	ut_ad(space->crypt_data != NULL && space->crypt_data->is_encrypted());
+	ut_ad(space->n_pending_ops > 0);
+
 	bool encrypted = fil_space_decrypt(
-				fil_space_get_crypt_data(space),
+				space->crypt_data,
 				tmp_frame,
-				page_size,
+				size,
 				src_frame,
 				&err);
 
@@ -894,7 +884,7 @@ fil_space_decrypt(
 			*decrypted = true;
 			/* Copy the decrypted page back to page buffer, not
 			really any other options. */
-			memcpy(src_frame, tmp_frame, page_size);
+			memcpy(src_frame, tmp_frame, size);
 		}
 
 		res = src_frame;
@@ -905,14 +895,15 @@ fil_space_decrypt(
 
 /******************************************************************
 Calculate post encryption checksum
+@param[in]	zip_size	zip_size or 0
+@param[in]	dst_frame	Block where checksum is calculated
 @return page checksum or BUF_NO_CHECKSUM_MAGIC
 not needed. */
 UNIV_INTERN
 ulint
 fil_crypt_calculate_checksum(
-/*=========================*/
-	ulint	zip_size,	/*!< in: zip_size or 0 */
-	byte*	dst_frame)	/*!< in: page where to calculate */
+	ulint		zip_size,
+	const byte*	dst_frame)
 {
 	ib_uint32_t checksum = 0;
 	srv_checksum_algorithm_t algorithm =
@@ -951,6 +942,7 @@ metadata (this is strong indication that tablespace is encrypted).
 Function also verifies that traditional checksum does not match
 calculated checksum as if it does page could be valid unencrypted,
 encrypted, or corrupted.
+
 @param[in]	page		Page to verify
 @param[in]	zip_size	zip size
 @param[in]	space		Tablespace
@@ -1043,7 +1035,6 @@ fil_space_verify_crypt_checksum(
 	assume that a nonzero value means that the page is encrypted.
 	Therefore we must validate the page both as encrypted and unencrypted
 	when FIL_PAGE_FILE_FLUSH_LSN does not contain 0.
-
 	*/
 
 	ulint checksum1 = mach_read_from_4(
@@ -1090,12 +1081,13 @@ struct key_state_t {
 };
 
 /***********************************************************************
-Copy global key state */
+Copy global key state
+@param[in,out]	new_state	key state
+@param[in]	crypt_data	crypt data */
 static void
 fil_crypt_get_key_state(
-/*====================*/
-	key_state_t*		new_state,	/*!< out: key state */
-	fil_space_crypt_t*	crypt_data)	/*!< in, out: crypt_data */
+	key_state_t*			new_state,
+	fil_space_crypt_t*		crypt_data)
 {
 	if (srv_encrypt_tables) {
 		new_state->key_version = crypt_data->key_get_latest_version();
@@ -1110,15 +1102,17 @@ fil_crypt_get_key_state(
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
+@param[in]	encrypt_mode		Encryption mode
+@param[in]	key_version		Current key version
+@param[in]	latest_key_version	Latest key version
+@param[in]	rotate_key_age		when to rotate
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
-/*=====================*/
-	fil_encryption_t        encrypt_mode,           /*!< in: Encryption
-							mode */
-	uint			key_version,		/*!< in: Key version */
-	uint			latest_key_version,	/*!< in: Latest key version */
-	uint			rotate_key_age)		/*!< in: When to rotate */
+	fil_encryption_t	encrypt_mode,
+	uint			key_version,
+	uint			latest_key_version,
+	uint			rotate_key_age)
 {
 	if (key_version == ENCRYPTION_KEY_VERSION_INVALID) {
 		return false;
@@ -1131,7 +1125,7 @@ fil_crypt_needs_rotation(
 	}
 
 	if (latest_key_version == 0 && key_version != 0) {
-		if (encrypt_mode == FIL_SPACE_ENCRYPTION_DEFAULT) {
+		if (encrypt_mode == FIL_ENCRYPTION_DEFAULT) {
 			/* this is rotation encrypted => unencrypted */
 			return true;
 		}
@@ -1148,59 +1142,34 @@ fil_crypt_needs_rotation(
 }
 
 /***********************************************************************
-Check if a space is closing (i.e just before drop)
-@return true if space is closing, false if not. */
-UNIV_INTERN
-bool
-fil_crypt_is_closing(
-/*=================*/
-	ulint space)	/*!< in: FIL space id */
-{
-	bool closing=true;
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
-
-	if (crypt_data) {
-		closing = crypt_data->is_closing(false);
-	}
-
-	return closing;
-}
-
-/***********************************************************************
 Start encrypting a space
-@return true if a pending op (fil_inc_pending_ops/fil_decr_pending_ops) is held
-*/
+@param[in,out]		space		Tablespace
+@return true if a recheck is needed */
 static
 bool
 fil_crypt_start_encrypting_space(
-/*=============================*/
-	ulint	space,	/*!< in: FIL space id */
-	bool*	recheck)/*!< out: true if recheck needed */
+	fil_space_t*	space)
 {
-
-	/* we have a pending op when entering function */
-	bool pending_op = true;
-
+	bool recheck = false;
 	mutex_enter(&fil_crypt_threads_mutex);
 
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
-	ibool page_encrypted = (crypt_data != NULL);
+	fil_space_crypt_t *crypt_data = space->crypt_data;
 
-	/*If spage is not encrypted and encryption is not enabled, then
+	/* If space is not encrypted and encryption is not enabled, then
 	do not continue encrypting the space. */
-	if (!page_encrypted && !srv_encrypt_tables) {
+	if (!crypt_data && !srv_encrypt_tables) {
 		mutex_exit(&fil_crypt_threads_mutex);
-		return pending_op;
+		return false;
 	}
 
 	if (crypt_data != NULL || fil_crypt_start_converting) {
 		/* someone beat us to it */
 		if (fil_crypt_start_converting) {
-			*recheck = true;
+			recheck = true;
 		}
 
 		mutex_exit(&fil_crypt_threads_mutex);
-		return pending_op;
+		return recheck;
 	}
 
 	/* NOTE: we need to write and flush page 0 before publishing
@@ -1209,10 +1178,11 @@ fil_crypt_start_encrypting_space(
 	* crypt data in page 0 */
 
 	/* 1 - create crypt data */
-	crypt_data = fil_space_create_crypt_data(FIL_SPACE_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+	crypt_data = fil_space_create_crypt_data(FIL_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+
 	if (crypt_data == NULL) {
 		mutex_exit(&fil_crypt_threads_mutex);
-		return pending_op;
+		return false;
 	}
 
 	crypt_data->type = CRYPT_SCHEME_UNENCRYPTED;
@@ -1230,87 +1200,44 @@ fil_crypt_start_encrypting_space(
 
 	do
 	{
-		if (fil_crypt_is_closing(space) ||
-			fil_space_found_by_id(space) == NULL) {
-			break;
-		}
-
 		mtr_t mtr;
 		mtr_start(&mtr);
 
 		/* 2 - get page 0 */
-		ulint offset = 0;
-		ulint zip_size = fil_space_get_zip_size(space);
-		buf_block_t* block = buf_page_get_gen(space, zip_size, offset,
+		ulint zip_size = fsp_flags_get_zip_size(space->flags);
+		buf_block_t* block = buf_page_get_gen(space->id, zip_size, 0,
 						      RW_X_LATCH,
 						      NULL,
 						      BUF_GET,
 						      __FILE__, __LINE__,
 						      &mtr);
 
-		if (fil_crypt_is_closing(space) ||
-		    fil_space_found_by_id(space) == NULL) {
-			mtr_commit(&mtr);
-			break;
-		}
 
-		/* 3 - compute location to store crypt data */
+		/* 3 - write crypt data to page 0 */
 		byte* frame = buf_block_get_frame(block);
-		ulint maxsize;
-		ut_ad(crypt_data);
-		crypt_data->page0_offset =
-			fsp_header_get_crypt_offset(zip_size, &maxsize);
-
-		/* 4 - write crypt data to page 0 */
-		fil_space_write_crypt_data_low(crypt_data,
-					       CRYPT_SCHEME_1,
-					       frame,
-					       crypt_data->page0_offset,
-					       maxsize, &mtr);
+		crypt_data->type = CRYPT_SCHEME_1;
+		crypt_data->write_page0(frame, &mtr);
 
-		mtr_commit(&mtr);
 
-		if (fil_crypt_is_closing(space) ||
-		    fil_space_found_by_id(space) == NULL) {
-			break;
-		}
+		mtr_commit(&mtr);
 
 		/* record lsn of update */
 		lsn_t end_lsn = mtr.end_lsn;
 
 		/* 4 - sync tablespace before publishing crypt data */
 
-		/* release "lock" while syncing */
-		fil_decr_pending_ops(space);
-		pending_op = false;
-
 		bool success = false;
-		ulint n_pages = 0;
 		ulint sum_pages = 0;
+
 		do {
+			ulint n_pages = 0;
 			success = buf_flush_list(ULINT_MAX, end_lsn, &n_pages);
 			buf_flush_wait_batch_end(NULL, BUF_FLUSH_LIST);
 			sum_pages += n_pages;
-		} while (!success &&
-			 !fil_crypt_is_closing(space) &&
-			 !fil_space_found_by_id(space));
-
-		/* try to reacquire pending op */
-		if (fil_inc_pending_ops(space, true)) {
-			break;
-		}
-
-		/* pending op reacquired! */
-		pending_op = true;
-
-		if (fil_crypt_is_closing(space) ||
-		    fil_space_found_by_id(space) == NULL) {
-			break;
-		}
+		} while (!success);
 
 		/* 5 - publish crypt data */
 		mutex_enter(&fil_crypt_threads_mutex);
-		ut_ad(crypt_data);
 		mutex_enter(&crypt_data->mutex);
 		crypt_data->type = CRYPT_SCHEME_1;
 		ut_a(crypt_data->rotate_state.active_threads == 1);
@@ -1321,10 +1248,9 @@ fil_crypt_start_encrypting_space(
 		mutex_exit(&crypt_data->mutex);
 		mutex_exit(&fil_crypt_threads_mutex);
 
-		return pending_op;
+		return recheck;
 	} while (0);
 
-	ut_ad(crypt_data);
 	mutex_enter(&crypt_data->mutex);
 	ut_a(crypt_data->rotate_state.active_threads == 1);
 	crypt_data->rotate_state.active_threads = 0;
@@ -1334,7 +1260,7 @@ fil_crypt_start_encrypting_space(
 	fil_crypt_start_converting = false;
 	mutex_exit(&fil_crypt_threads_mutex);
 
-	return pending_op;
+	return recheck;
 }
 
 /** State of a rotation thread */
@@ -1348,7 +1274,7 @@ struct rotate_thread_t {
 
 	uint thread_no;
 	bool first;		    /*!< is position before first space */
-	ulint space;		    /*!< current space */
+	fil_space_t* space;	    /*!< current space or NULL */
 	ulint offset;		    /*!< current offset */
 	ulint batch;		    /*!< #pages to rotate */
 	uint  min_key_version_found;/*!< min key version found but not rotated */
@@ -1383,54 +1309,41 @@ struct rotate_thread_t {
 
 /***********************************************************************
 Check if space needs rotation given a key_state
+@param[in,out]		state		Key rotation state
+@param[in,out]		key_state	Key state
+@param[in,out]		recheck		needs recheck ?
 @return true if space needs key rotation */
 static
 bool
 fil_crypt_space_needs_rotation(
-/*===========================*/
-	rotate_thread_t*	state,		/*!< in: Key rotation state */
-	key_state_t*		key_state,	/*!< in: Key state */
-	bool*			recheck)	/*!< out: needs recheck ? */
+	rotate_thread_t*	state,
+	key_state_t*		key_state,
+	bool*			recheck)
 {
-	ulint space = state->space;
-
-	/* Make sure that tablespace is found and it is normal tablespace */
-	if (fil_space_found_by_id(space) == NULL ||
-		fil_space_get_type(space) != FIL_TABLESPACE) {
-		return false;
-	}
+	fil_space_t* space = state->space;
 
-	if (fil_inc_pending_ops(space, true)) {
-		/* tablespace being dropped */
+	/* Make sure that tablespace is normal tablespace */
+	if (space->purpose != FIL_TABLESPACE) {
 		return false;
 	}
 
-	/* keep track of if we have pending op */
-	bool pending_op = true;
+	ut_ad(space->n_pending_ops > 0);
 
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
+	fil_space_crypt_t *crypt_data = space->crypt_data;
 
 	if (crypt_data == NULL) {
 		/**
 		* space has no crypt data
 		*   start encrypting it...
 		*/
-		pending_op = fil_crypt_start_encrypting_space(space, recheck);
-
-		crypt_data = fil_space_get_crypt_data(space);
+		*recheck = fil_crypt_start_encrypting_space(space);
+		crypt_data = space->crypt_data;
 
 		if (crypt_data == NULL) {
-			if (pending_op) {
-				fil_decr_pending_ops(space);
-			}
 			return false;
 		}
 
 		crypt_data->key_get_latest_version();
-
-		if (!crypt_data->is_key_found()) {
-			return false;
-		}
 	}
 
 	/* If used key_id is not found from encryption plugin we can't
@@ -1450,7 +1363,7 @@ fil_crypt_space_needs_rotation(
 		}
 
 		/* prevent threads from starting to rotate space */
-		if (crypt_data->is_closing(true)) {
+		if (space->is_stopping()) {
 			break;
 		}
 
@@ -1474,12 +1387,14 @@ fil_crypt_space_needs_rotation(
 			key_state->key_version, key_state->rotate_key_age);
 
 		crypt_data->rotate_state.scrubbing.is_active =
-			btr_scrub_start_space(space, &state->scrub_data);
+			btr_scrub_start_space(space->id, &state->scrub_data);
 
 		time_t diff = time(0) - crypt_data->rotate_state.scrubbing.
 			last_scrub_completed;
 
 		bool need_scrubbing =
+			(srv_background_scrub_data_uncompressed ||
+			 srv_background_scrub_data_compressed) &&
 			crypt_data->rotate_state.scrubbing.is_active
 			&& diff >= 0
 			&& ulint(diff) >= srv_background_scrub_data_interval;
@@ -1489,25 +1404,22 @@ fil_crypt_space_needs_rotation(
 		}
 
 		mutex_exit(&crypt_data->mutex);
-		/* NOTE! fil_decr_pending_ops is performed outside */
+
 		return true;
 	} while (0);
 
 	mutex_exit(&crypt_data->mutex);
 
-	if (pending_op) {
-		fil_decr_pending_ops(space);
-	}
 
 	return false;
 }
 
 /***********************************************************************
-Update global statistics with thread statistics */
+Update global statistics with thread statistics
+@param[in,out]	state		key rotation statistics */
 static void
 fil_crypt_update_total_stat(
-/*========================*/
-	rotate_thread_t *state)	/*!< in: Key rotation status */
+	rotate_thread_t *state)
 {
 	mutex_enter(&crypt_stat_mutex);
 	crypt_stat.pages_read_from_cache +=
@@ -1531,15 +1443,19 @@ fil_crypt_update_total_stat(
 /***********************************************************************
 Allocate iops to thread from global setting,
 used before starting to rotate a space.
+@param[in,out]		state		Rotation state
 @return true if allocation succeeded, false if failed */
 static
 bool
 fil_crypt_alloc_iops(
-/*=================*/
-	rotate_thread_t *state)	/*!< in: Key rotation status */
+	rotate_thread_t *state)
 {
 	ut_ad(state->allocated_iops == 0);
 
+	/* We have not yet selected the space to rotate, thus
+	state might not contain space and we can't check
+	its status yet. */
+
 	uint max_iops = state->estimated_max_iops;
 	mutex_enter(&fil_crypt_threads_mutex);
 
@@ -1565,12 +1481,12 @@ fil_crypt_alloc_iops(
 
 /***********************************************************************
 Reallocate iops to thread,
-used when inside a space */
+used when inside a space
+@param[in,out]		state		Rotation state */
 static
 void
 fil_crypt_realloc_iops(
-/*===================*/
-	rotate_thread_t *state)	/*!< in: Key rotation status */
+	rotate_thread_t *state)
 {
 	ut_a(state->allocated_iops > 0);
 
@@ -1579,13 +1495,12 @@ fil_crypt_realloc_iops(
 		uint avg_wait_time_us =
 			state->sum_waited_us / state->cnt_waited;
 
-#if DEBUG_KEYROTATION_THROTTLING
-		ib_logf(IB_LOG_LEVEL_INFO,
-			"thr_no: %u - update estimated_max_iops from %u to %u.",
+		DBUG_PRINT("ib_crypt",
+			("thr_no: %u - update estimated_max_iops from %u to %u.",
 			state->thread_no,
 			state->estimated_max_iops,
-			1000000 / avg_wait_time_us);
-#endif
+			1000000 / avg_wait_time_us));
+
 		if (avg_wait_time_us == 0) {
 			avg_wait_time_us = 1; // prevent division by zero
 		}
@@ -1594,12 +1509,11 @@ fil_crypt_realloc_iops(
 		state->cnt_waited = 0;
 		state->sum_waited_us = 0;
 	} else {
-#if DEBUG_KEYROTATION_THROTTLING
-		ib_logf(IB_LOG_LEVEL_INFO,
-			"thr_no: %u only waited %lu%% skip re-estimate.",
+
+		DBUG_PRINT("ib_crypt",
+			("thr_no: %u only waited %lu%% skip re-estimate.",
 			state->thread_no,
-			(100 * state->cnt_waited) / state->batch);
-#endif
+			(100 * state->cnt_waited) / state->batch));
 	}
 
 	if (state->estimated_max_iops <= state->allocated_iops) {
@@ -1625,8 +1539,9 @@ fil_crypt_realloc_iops(
 				state->allocated_iops ++;
 				n_fil_crypt_iops_allocated ++;
 			}
-			mutex_exit(&fil_crypt_threads_mutex);
+
 			os_event_set(fil_crypt_threads_event);
+			mutex_exit(&fil_crypt_threads_mutex);
 		}
 	} else {
 		/* see if there are more to get */
@@ -1643,13 +1558,13 @@ fil_crypt_realloc_iops(
 			}
 			n_fil_crypt_iops_allocated += extra;
 			state->allocated_iops += extra;
-#if DEBUG_KEYROTATION_THROTTLING
-			ib_logf(IB_LOG_LEVEL_INFO,
-				"thr_no: %u increased iops from %u to %u.",
+
+			DBUG_PRINT("ib_crypt",
+				("thr_no: %u increased iops from %u to %u.",
 				state->thread_no,
 				state->allocated_iops - extra,
-				state->allocated_iops);
-#endif
+				state->allocated_iops));
+
 		}
 		mutex_exit(&fil_crypt_threads_mutex);
 	}
@@ -1658,12 +1573,12 @@ fil_crypt_realloc_iops(
 }
 
 /***********************************************************************
-Return allocated iops to global */
+Return allocated iops to global
+@param[in,out]		state		Rotation state */
 static
 void
 fil_crypt_return_iops(
-/*==================*/
-	rotate_thread_t *state)	/*!< in: Key rotation status */
+	rotate_thread_t *state)
 {
 	if (state->allocated_iops > 0) {
 		uint iops = state->allocated_iops;
@@ -1676,25 +1591,27 @@ fil_crypt_return_iops(
 			ut_ad(0);
 			iops = 0;
 		}
+
 		n_fil_crypt_iops_allocated -= iops;
-		mutex_exit(&fil_crypt_threads_mutex);
 		state->allocated_iops = 0;
 		os_event_set(fil_crypt_threads_event);
+		mutex_exit(&fil_crypt_threads_mutex);
 	}
 
 	fil_crypt_update_total_stat(state);
 }
 
 /***********************************************************************
-Search for a space needing rotation */
-UNIV_INTERN
+Search for a space needing rotation
+@param[in,out]		key_state		Key state
+@param[in,out]		state			Rotation state
+@param[in,out]		recheck			recheck ? */
+static
 bool
 fil_crypt_find_space_to_rotate(
-/*===========================*/
-	key_state_t*		key_state,	/*!< in: Key state */
-	rotate_thread_t*	state,		/*!< in: Key rotation state */
-	bool*			recheck)	/*!< out: true if recheck
-						needed */
+	key_state_t*		key_state,
+	rotate_thread_t*	state,
+	bool*			recheck)
 {
 	/* we need iops to start rotating */
 	while (!state->should_shutdown() && !fil_crypt_alloc_iops(state)) {
@@ -1703,30 +1620,44 @@ fil_crypt_find_space_to_rotate(
 	}
 
 	if (state->should_shutdown()) {
+		if (state->space) {
+			fil_space_release(state->space);
+			state->space = NULL;
+		}
 		return false;
 	}
 
 	if (state->first) {
 		state->first = false;
-		state->space = fil_get_first_space_safe();
-	} else {
-		state->space = fil_get_next_space_safe(state->space);
+		if (state->space) {
+			fil_space_release(state->space);
+		}
+		state->space = NULL;
 	}
 
-	while (!state->should_shutdown() && state->space != ULINT_UNDEFINED) {
-		fil_space_t* space = fil_space_found_by_id(state->space);
+	/* If key rotation is enabled (default) we iterate all tablespaces.
+	If key rotation is not enabled we iterate only the tablespaces
+	added to keyrotation list. */
+	if (srv_fil_crypt_rotate_key_age) {
+		state->space = fil_space_next(state->space);
+	} else {
+		state->space = fil_space_keyrotate_next(state->space);
+	}
 
-		if (space) {
-			if (fil_crypt_space_needs_rotation(state, key_state, recheck)) {
-				ut_ad(key_state->key_id);
-				/* init state->min_key_version_found before
-				* starting on a space */
-				state->min_key_version_found = key_state->key_version;
-				return true;
-			}
+	while (!state->should_shutdown() && 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
+			* starting on a space */
+			state->min_key_version_found = key_state->key_version;
+			return true;
 		}
 
-		state->space = fil_get_next_space_safe(state->space);
+		if (srv_fil_crypt_rotate_key_age) {
+			state->space = fil_space_next(state->space);
+		} else {
+			state->space = fil_space_keyrotate_next(state->space);
+		}
 	}
 
 	/* if we didn't find any space return iops */
@@ -1737,16 +1668,16 @@ fil_crypt_find_space_to_rotate(
 }
 
 /***********************************************************************
-Start rotating a space */
+Start rotating a space
+@param[in]	key_state		Key state
+@param[in,out]	state			Rotation state */
 static
 void
 fil_crypt_start_rotate_space(
-/*=========================*/
-	const key_state_t*	key_state,	/*!< in: Key state */
-	rotate_thread_t*	state)		/*!< in: Key rotation state */
+	const key_state_t*	key_state,
+	rotate_thread_t*	state)
 {
-	ulint space = state->space;
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
+	fil_space_crypt_t *crypt_data = state->space->crypt_data;
 
 	ut_ad(crypt_data);
 	mutex_enter(&crypt_data->mutex);
@@ -1757,8 +1688,9 @@ fil_crypt_start_rotate_space(
 		crypt_data->rotate_state.next_offset = 1; // skip page 0
 		/* no need to rotate beyond current max
 		* if space extends, it will be encrypted with newer version */
-		crypt_data->rotate_state.max_offset = fil_space_get_size(space);
-
+		/* FIXME: max_offset could be removed and instead
+		space->size consulted.*/
+		crypt_data->rotate_state.max_offset = state->space->size;
 		crypt_data->rotate_state.end_lsn = 0;
 		crypt_data->rotate_state.min_key_version_found =
 			key_state->key_version;
@@ -1786,26 +1718,34 @@ fil_crypt_start_rotate_space(
 
 /***********************************************************************
 Search for batch of pages needing rotation
+@param[in]	key_state		Key state
+@param[in,out]	state			Rotation state
 @return true if page needing key rotation found, false if not found */
 static
 bool
 fil_crypt_find_page_to_rotate(
-/*==========================*/
-	const key_state_t*	key_state,	/*!< in: Key state */
-	rotate_thread_t*	state)		/*!< in: Key rotation state */
+	const key_state_t*	key_state,
+	rotate_thread_t*	state)
 {
 	ulint batch = srv_alloc_time * state->allocated_iops;
-	ulint space = state->space;
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
+	fil_space_t* space = state->space;
+
+	ut_ad(!space || space->n_pending_ops > 0);
+
+	/* If space is marked to be dropped stop rotation. */
+	if (!space || space->is_stopping()) {
+		return false;
+	}
+
+	fil_space_crypt_t *crypt_data = space->crypt_data;
 
 	/* Space might already be dropped */
 	if (crypt_data) {
 		mutex_enter(&crypt_data->mutex);
 		ut_ad(key_state->key_id == crypt_data->key_id);
 
-		if (!crypt_data->is_closing(true) &&
-			crypt_data->rotate_state.next_offset <
-			crypt_data->rotate_state.max_offset) {
+		if (crypt_data->rotate_state.next_offset <
+		    crypt_data->rotate_state.max_offset) {
 
 			state->offset = crypt_data->rotate_state.next_offset;
 			ulint remaining = crypt_data->rotate_state.max_offset -
@@ -1830,59 +1770,47 @@ fil_crypt_find_page_to_rotate(
 
 /***********************************************************************
 Check if a page is uninitialized (doesn't need to be rotated)
-@return true if page is uninitialized, false if not.*/
-static
+@param[in]	frame		Page to check
+@param[in]	zip_size	zip_size or 0
+@return true if page is uninitialized, false if not. */
+static inline
 bool
 fil_crypt_is_page_uninitialized(
-/*============================*/
-	const byte	*frame, 	/*!< in: Page */
-	uint		zip_size)	/*!< in: compressed size if
-					row_format compressed */
+	const byte	*frame,
+	uint		zip_size)
 {
-	if (zip_size) {
-		ulint stored_checksum = mach_read_from_4(
-			frame + FIL_PAGE_SPACE_OR_CHKSUM);
-		/* empty pages aren't encrypted */
-		if (stored_checksum == 0) {
-			return true;
-		}
-	} else {
-		ulint size = UNIV_PAGE_SIZE;
-		ulint checksum_field1 = mach_read_from_4(
-			frame + FIL_PAGE_SPACE_OR_CHKSUM);
-		ulint checksum_field2 = mach_read_from_4(
-			frame + size - FIL_PAGE_END_LSN_OLD_CHKSUM);
-		/* empty pages are not encrypted */
-		if (checksum_field1 == 0 && checksum_field2 == 0
-		    && mach_read_from_4(frame + FIL_PAGE_LSN) == 0) {
-			return true;
-		}
-	}
-	return false;
+	return (buf_page_is_zeroes(frame, zip_size));
 }
 
-#define fil_crypt_get_page_throttle(state,space,zip_size,offset,mtr,sleeptime_ms) \
-	fil_crypt_get_page_throttle_func(state, space, zip_size, offset, mtr, \
+#define fil_crypt_get_page_throttle(state,offset,mtr,sleeptime_ms) \
+	fil_crypt_get_page_throttle_func(state, offset, mtr, \
 					 sleeptime_ms, __FILE__, __LINE__)
 
 /***********************************************************************
 Get a page and compute sleep time
-@return page */
+@param[in,out]		state		Rotation state
+@param[in]		zip_size	compressed size or 0
+@param[in]		offset		Page offset
+@param[in,out]		mtr		Minitransaction
+@param[out]		sleeptime_ms	Sleep time
+@param[in]		file		File where called
+@param[in]		line		Line where called
+@return page or NULL*/
 static
 buf_block_t*
 fil_crypt_get_page_throttle_func(
-/*=============================*/
-	rotate_thread_t*	state,		/*!< in/out: Key rotation state */
-	ulint			space,		/*!< in: FIL space id */
-	uint 			zip_size,	/*!< in: compressed size if
-						row_format compressed */
-	ulint 			offset,		/*!< in: page offsett */
-	mtr_t*			mtr,		/*!< in/out: minitransaction */
-	ulint*			sleeptime_ms,	/*!< out: sleep time */
-	const char*		file,		/*!< in: file name */
-	ulint 			line)		/*!< in: file line */
+	rotate_thread_t*	state,
+	ulint 			offset,
+	mtr_t*			mtr,
+	ulint*			sleeptime_ms,
+	const char*		file,
+	ulint 			line)
 {
-	buf_block_t* block = buf_page_try_get_func(space, offset, RW_X_LATCH,
+	fil_space_t* space = state->space;
+	ulint zip_size = fsp_flags_get_zip_size(space->flags);
+	ut_ad(space->n_pending_ops > 0);
+
+	buf_block_t* block = buf_page_try_get_func(space->id, offset, RW_X_LATCH,
 						   true,
 						   file, line, mtr);
 	if (block != NULL) {
@@ -1893,16 +1821,14 @@ fil_crypt_get_page_throttle_func(
 
 	/* Before reading from tablespace we need to make sure that
 	tablespace exists and is not is just being dropped. */
-
-	if (fil_crypt_is_closing(space) ||
-		fil_space_found_by_id(space) == NULL) {
+	if (space->is_stopping()) {
 		return NULL;
 	}
 
 	state->crypt_stat.pages_read_from_disk++;
 
 	ullint start = ut_time_us(NULL);
-	block = buf_page_get_gen(space, zip_size, offset,
+	block = buf_page_get_gen(space->id, zip_size, offset,
 				 RW_X_LATCH,
 				 NULL, BUF_GET_POSSIBLY_FREED,
 				 file, line, mtr);
@@ -1928,6 +1854,7 @@ fil_crypt_get_page_throttle_func(
 	}
 
 	*sleeptime_ms += add_sleeptime_ms;
+
 	return block;
 }
 
@@ -1937,27 +1864,35 @@ Get block and allocation status
 
 note: innodb locks fil_space_latch and then block when allocating page
 but locks block and then fil_space_latch when freeing page.
-@return block
+
+@param[in,out]		state		Rotation state
+@param[in]		zip_size	Compressed size or 0
+@param[in]		offset		Page offset
+@param[in,out]		mtr		Minitransaction
+@param[out]		allocation_status Allocation status
+@param[out]		sleeptime_ms	Sleep time
+@return block or NULL
 */
 static
 buf_block_t*
 btr_scrub_get_block_and_allocation_status(
-/*======================================*/
-	rotate_thread_t*	state,		/*!< in/out: Key rotation state */
-	ulint			space,		/*!< in: FIL space id */
-	uint 			zip_size,	/*!< in: compressed size if
-						row_format compressed */
-	ulint 			offset,		/*!< in: page offsett */
-	mtr_t*			mtr,		/*!< in/out: minitransaction
-						*/
+	rotate_thread_t*	state,
+	uint 			zip_size,
+	ulint 			offset,
+	mtr_t*			mtr,
 	btr_scrub_page_allocation_status_t *allocation_status,
-						/*!< in/out: allocation status */
-	ulint*			sleeptime_ms)	/*!< out: sleep time */
+	ulint*			sleeptime_ms)
 {
 	mtr_t local_mtr;
 	buf_block_t *block = NULL;
+	fil_space_t* space = state->space;
+
+	ut_ad(space->n_pending_ops > 0);
+	ut_ad(zip_size == fsp_flags_get_zip_size(space->flags));
+
 	mtr_start(&local_mtr);
-	*allocation_status = fsp_page_is_free(space, offset, &local_mtr) ?
+
+	*allocation_status = fsp_page_is_free(space->id, offset, &local_mtr) ?
 		BTR_SCRUB_PAGE_FREE :
 		BTR_SCRUB_PAGE_ALLOCATED;
 
@@ -1965,7 +1900,6 @@ btr_scrub_get_block_and_allocation_status(
 		/* this is easy case, we lock fil_space_latch first and
 		then block */
 		block = fil_crypt_get_page_throttle(state,
-						    space, zip_size,
 						    offset, mtr,
 						    sleeptime_ms);
 		mtr_commit(&local_mtr);
@@ -1982,7 +1916,6 @@ btr_scrub_get_block_and_allocation_status(
 		*/
 
 		block = fil_crypt_get_page_throttle(state,
-						    space, zip_size,
 						    offset, mtr,
 						    sleeptime_ms);
 	}
@@ -1992,21 +1925,29 @@ btr_scrub_get_block_and_allocation_status(
 
 
 /***********************************************************************
-Rotate one page */
+Rotate one page
+@param[in,out]		key_state		Key state
+@param[in,out]		state			Rotation state */
 static
 void
 fil_crypt_rotate_page(
-/*==================*/
-	const key_state_t*	key_state,	/*!< in: Key state */
-	rotate_thread_t*	state)		/*!< in: Key rotation state */
+	const key_state_t*	key_state,
+	rotate_thread_t*	state)
 {
-	ulint space = state->space;
+	fil_space_t*space = state->space;
+	ulint space_id = space->id;
 	ulint offset = state->offset;
-	const uint zip_size = fil_space_get_zip_size(space);
+	const uint zip_size = fsp_flags_get_zip_size(space->flags);
 	ulint sleeptime_ms = 0;
+	fil_space_crypt_t *crypt_data = space->crypt_data;
+
+	ut_ad(space->n_pending_ops > 0);
 
-	/* check if tablespace is closing before reading page */
-	if (fil_crypt_is_closing(space) || fil_space_found_by_id(space) == NULL) {
+	/* In fil_crypt_thread where key rotation is done we have
+	acquired space and checked that this space is not yet
+	marked to be dropped. Similarly, in fil_crypt_find_page_to_rotate().
+	Check here also to give DROP TABLE or similar a change. */
+	if (space->is_stopping()) {
 		return;
 	}
 
@@ -2018,7 +1959,6 @@ fil_crypt_rotate_page(
 	mtr_t mtr;
 	mtr_start(&mtr);
 	buf_block_t* block = fil_crypt_get_page_throttle(state,
-							 space, zip_size,
 							 offset, &mtr,
 							 &sleeptime_ms);
 
@@ -2030,9 +1970,8 @@ fil_crypt_rotate_page(
 		uint kv =  block->page.key_version;
 
 		/* check if tablespace is closing after reading page */
-		if (!fil_crypt_is_closing(space)) {
+		if (space->is_stopping()) {
 			byte* frame = buf_block_get_frame(block);
-			fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
 
 			if (kv == 0 &&
 				fil_crypt_is_page_uninitialized(frame, zip_size)) {
@@ -2052,7 +1991,7 @@ fil_crypt_rotate_page(
 				/* force rotation by dummy updating page */
 				mlog_write_ulint(frame +
 					FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID,
-					space, MLOG_4BYTES, &mtr);
+					space_id, MLOG_4BYTES, &mtr);
 
 				/* update block */
 				block->page.key_version = key_state->key_version;
@@ -2085,7 +2024,7 @@ fil_crypt_rotate_page(
 			*/
 			btr_scrub_page_allocation_status_t allocated;
 			block = btr_scrub_get_block_and_allocation_status(
-				state, space, zip_size, offset, &mtr,
+				state, zip_size, offset, &mtr,
 				&allocated,
 				&sleeptime_ms);
 
@@ -2099,7 +2038,7 @@ fil_crypt_rotate_page(
 					/* we need to refetch it once more now that we have
 					* index locked */
 					block = btr_scrub_get_block_and_allocation_status(
-						state, space, zip_size, offset, &mtr,
+						state, zip_size, offset, &mtr,
 						&allocated,
 						&sleeptime_ms);
 
@@ -2130,7 +2069,6 @@ fil_crypt_rotate_page(
 		if (needs_scrubbing == BTR_SCRUB_TURNED_OFF) {
 			/* if we just detected that scrubbing was turned off
 			* update global state to reflect this */
-			fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
 			ut_ad(crypt_data);
 			mutex_enter(&crypt_data->mutex);
 			crypt_data->rotate_state.scrubbing.is_active = false;
@@ -2158,17 +2096,20 @@ fil_crypt_rotate_page(
 }
 
 /***********************************************************************
-Rotate a batch of pages */
+Rotate a batch of pages
+@param[in,out]		key_state		Key state
+@param[in,out]		state			Rotation state */
 static
 void
 fil_crypt_rotate_pages(
-/*===================*/
-	const key_state_t*	key_state,	/*!< in: Key state */
-	rotate_thread_t*	state)		/*!< in: Key rotation state */
+	const key_state_t*	key_state,
+	rotate_thread_t*	state)
 {
-	ulint space = state->space;
+	ulint space = state->space->id;
 	ulint end = state->offset + state->batch;
 
+	ut_ad(state->space->n_pending_ops > 0);
+
 	for (; state->offset < end; state->offset++) {
 
 		/* we can't rotate pages in dblwr buffer as
@@ -2189,20 +2130,23 @@ fil_crypt_rotate_pages(
 }
 
 /***********************************************************************
-Flush rotated pages and then update page 0 */
+Flush rotated pages and then update page 0
+
+@param[in,out]		state	rotation state */
 static
 void
 fil_crypt_flush_space(
-/*==================*/
-	rotate_thread_t*	state,	/*!< in: Key rotation state */
-	ulint			space)	/*!< in: FIL space id */
+	rotate_thread_t*	state)
 {
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
+	fil_space_t* space = state->space;
+	fil_space_crypt_t *crypt_data = space->crypt_data;
+
+	ut_ad(space->n_pending_ops > 0);
 
 	/* flush tablespace pages so that there are no pages left with old key */
 	lsn_t end_lsn = crypt_data->rotate_state.end_lsn;
 
-	if (end_lsn > 0 && !fil_crypt_is_closing(space)) {
+	if (end_lsn > 0 && !space->is_stopping()) {
 		bool success = false;
 		ulint n_pages = 0;
 		ulint sum_pages = 0;
@@ -2212,7 +2156,7 @@ fil_crypt_flush_space(
 			success = buf_flush_list(ULINT_MAX, end_lsn, &n_pages);
 			buf_flush_wait_batch_end(NULL, BUF_FLUSH_LIST);
 			sum_pages += n_pages;
-		} while (!success && !fil_crypt_is_closing(space));
+		} while (!success && !space->is_stopping());
 
 		ullint end = ut_time_us(NULL);
 
@@ -2230,40 +2174,38 @@ fil_crypt_flush_space(
 	}
 
 	/* update page 0 */
-	if (!fil_crypt_is_closing(space)) {
-		mtr_t mtr;
-		mtr_start(&mtr);
-		ulint offset = 0; // page 0
-		const uint zip_size = fil_space_get_zip_size(space);
-		buf_block_t* block = buf_page_get_gen(space, zip_size, offset,
-						      RW_X_LATCH, NULL, BUF_GET,
-						      __FILE__, __LINE__, &mtr);
-		byte* frame = buf_block_get_frame(block);
-		ulint maxsize;
-		crypt_data->page0_offset =
-			fsp_header_get_crypt_offset(zip_size, &maxsize);
+	mtr_t mtr;
+	mtr_start(&mtr);
 
-		fil_space_write_crypt_data(space, frame,
-					   crypt_data->page0_offset,
-					   ULINT_MAX, &mtr);
-		mtr_commit(&mtr);
-	}
+	const uint zip_size = fsp_flags_get_zip_size(state->space->flags);
+
+	buf_block_t* block = buf_page_get_gen(space->id, zip_size, 0,
+				RW_X_LATCH, NULL, BUF_GET,
+				__FILE__, __LINE__, &mtr);
+	byte* frame = buf_block_get_frame(block);
+
+	crypt_data->write_page0(frame, &mtr);
+
+	mtr_commit(&mtr);
 }
 
 /***********************************************************************
-Complete rotating a space */
+Complete rotating a space
+@param[in,out]		key_state		Key state
+@param[in,out]		state			Rotation state */
 static
 void
 fil_crypt_complete_rotate_space(
-/*============================*/
-	const key_state_t*	key_state,	/*!< in: Key state */
-	rotate_thread_t*	state)		/*!< in: Key rotation state */
+	const key_state_t*	key_state,
+	rotate_thread_t*	state)
 {
-	ulint space = state->space;
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space);
+	fil_space_crypt_t *crypt_data = state->space->crypt_data;
+
+	ut_ad(crypt_data);
+	ut_ad(state->space->n_pending_ops > 0);
 
 	/* Space might already be dropped */
-	if (crypt_data != NULL && !crypt_data->is_closing(false)) {
+	if (!state->space->is_stopping()) {
 		mutex_enter(&crypt_data->mutex);
 
 		/**
@@ -2321,9 +2263,8 @@ fil_crypt_complete_rotate_space(
 		}
 
 		if (should_flush) {
-			fil_crypt_flush_space(state, space);
+			fil_crypt_flush_space(state);
 
-			ut_ad(crypt_data);
 			mutex_enter(&crypt_data->mutex);
 			crypt_data->rotate_state.flushing = false;
 			mutex_exit(&crypt_data->mutex);
@@ -2346,8 +2287,8 @@ DECLARE_THREAD(fil_crypt_thread)(
 	mutex_enter(&fil_crypt_threads_mutex);
 	uint thread_no = srv_n_fil_crypt_threads_started;
 	srv_n_fil_crypt_threads_started++;
-	mutex_exit(&fil_crypt_threads_mutex);
 	os_event_set(fil_crypt_event); /* signal that we started */
+	mutex_exit(&fil_crypt_threads_mutex);
 
 	/* state of this thread */
 	rotate_thread_t thr(thread_no);
@@ -2367,6 +2308,7 @@ DECLARE_THREAD(fil_crypt_thread)(
 			* i.e either new key version of change or
 			* new rotate_key_age */
 			os_event_reset(fil_crypt_threads_event);
+
 			if (os_event_wait_time(fil_crypt_threads_event, 1000000) == 0) {
 				break;
 			}
@@ -2380,9 +2322,12 @@ DECLARE_THREAD(fil_crypt_thread)(
 
 			time_t waited = time(0) - wait_start;
 
+			/* Break if we have waited the background scrub
+			internal and background scrubbing is enabled */
 			if (waited >= 0
-			    && ulint(waited)
-			    >= srv_background_scrub_data_check_interval) {
+			    && ulint(waited) >= srv_background_scrub_data_check_interval
+			    && (srv_background_scrub_data_uncompressed
+			        || srv_background_scrub_data_compressed)) {
 				break;
 			}
 		}
@@ -2397,29 +2342,32 @@ DECLARE_THREAD(fil_crypt_thread)(
 			/* we found a space to rotate */
 			fil_crypt_start_rotate_space(&new_state, &thr);
 
-			/* decrement pending ops that was incremented in
-			* fil_crypt_space_needs_rotation
-			* (called from fil_crypt_find_space_to_rotate),
-			* this makes sure that tablespace won't be dropped
-			* just after we decided to start processing it. */
-			fil_decr_pending_ops(thr.space);
-
 			/* iterate all pages (cooperativly with other threads) */
-			while (!thr.should_shutdown() &&
+			while (!thr.should_shutdown() && thr.space &&
 			       fil_crypt_find_page_to_rotate(&new_state, &thr)) {
 
 				/* rotate a (set) of pages */
 				fil_crypt_rotate_pages(&new_state, &thr);
 
+				/* If space is marked as stopping, release
+				space and stop rotation. */
+				if (thr.space->is_stopping()) {
+					fil_space_release(thr.space);
+					thr.space = NULL;
+					break;
+				}
+
 				/* realloc iops */
 				fil_crypt_realloc_iops(&thr);
 			}
 
 			/* complete rotation */
-			fil_crypt_complete_rotate_space(&new_state, &thr);
+			if (thr.space) {
+				fil_crypt_complete_rotate_space(&new_state, &thr);
+			}
 
 			/* force key state refresh */
-			new_state.key_id= 0;
+			new_state.key_id = 0;
 
 			/* return iops */
 			fil_crypt_return_iops(&thr);
@@ -2429,10 +2377,16 @@ DECLARE_THREAD(fil_crypt_thread)(
 	/* return iops if shutting down */
 	fil_crypt_return_iops(&thr);
 
+	/* release current space if shutting down */
+	if (thr.space) {
+		fil_space_release(thr.space);
+		thr.space = NULL;
+	}
+
 	mutex_enter(&fil_crypt_threads_mutex);
 	srv_n_fil_crypt_threads_started--;
-	mutex_exit(&fil_crypt_threads_mutex);
 	os_event_set(fil_crypt_event); /* signal that we stopped */
+	mutex_exit(&fil_crypt_threads_mutex);
 
 	/* 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. */
@@ -2443,23 +2397,26 @@ DECLARE_THREAD(fil_crypt_thread)(
 }
 
 /*********************************************************************
-Adjust thread count for key rotation */
+Adjust thread count for key rotation
+@param[in]	enw_cnt		Number of threads to be used */
 UNIV_INTERN
 void
 fil_crypt_set_thread_cnt(
-/*=====================*/
-	uint	new_cnt)	/*!< in: New key rotation thread count */
+	const uint	new_cnt)
 {
 	if (!fil_crypt_threads_inited) {
 		fil_crypt_threads_init();
 	}
 
+	mutex_enter(&fil_crypt_threads_mutex);
+
 	if (new_cnt > srv_n_fil_crypt_threads) {
 		uint add = new_cnt - srv_n_fil_crypt_threads;
 		srv_n_fil_crypt_threads = new_cnt;
 		for (uint i = 0; i < add; i++) {
 			os_thread_id_t rotation_thread_id;
 			os_thread_create(fil_crypt_thread, NULL, &rotation_thread_id);
+
 			ib_logf(IB_LOG_LEVEL_INFO,
 				"Creating #%d thread id %lu total threads %u.",
 				i+1, os_thread_pf(rotation_thread_id), new_cnt);
@@ -2469,6 +2426,8 @@ fil_crypt_set_thread_cnt(
 		os_event_set(fil_crypt_threads_event);
 	}
 
+	mutex_exit(&fil_crypt_threads_mutex);
+
 	while(srv_n_fil_crypt_threads_started != srv_n_fil_crypt_threads) {
 		os_event_reset(fil_crypt_event);
 		os_event_wait_time(fil_crypt_event, 1000000);
@@ -2476,39 +2435,39 @@ fil_crypt_set_thread_cnt(
 }
 
 /*********************************************************************
-Adjust max key age */
+Adjust max key age
+@param[in]	val		New max key age */
 UNIV_INTERN
 void
 fil_crypt_set_rotate_key_age(
-/*=========================*/
-	uint	val)	/*!< in: New max key age */
+	uint	val)
 {
 	srv_fil_crypt_rotate_key_age = val;
 	os_event_set(fil_crypt_threads_event);
 }
 
 /*********************************************************************
-Adjust rotation iops */
+Adjust rotation iops
+@param[in]	val		New max roation iops */
 UNIV_INTERN
 void
 fil_crypt_set_rotation_iops(
-/*========================*/
-	uint val)	/*!< in: New iops setting */
+	uint val)
 {
 	srv_n_fil_crypt_iops = val;
 	os_event_set(fil_crypt_threads_event);
 }
 
 /*********************************************************************
-Adjust encrypt tables */
+Adjust encrypt tables
+@param[in]	val		New setting for innodb-encrypt-tables */
 UNIV_INTERN
 void
 fil_crypt_set_encrypt_tables(
-/*=========================*/
-       uint val)       /*!< in: New srv_encrypt_tables setting */
+	uint val)
 {
-       srv_encrypt_tables = val;
-       os_event_set(fil_crypt_threads_event);
+	srv_encrypt_tables = val;
+	os_event_set(fil_crypt_threads_event);
 }
 
 /*********************************************************************
@@ -2516,7 +2475,6 @@ Init threads for key rotation */
 UNIV_INTERN
 void
 fil_crypt_threads_init()
-/*====================*/
 {
 	ut_ad(mutex_own(&fil_system->mutex));
 	if (!fil_crypt_threads_inited) {
@@ -2537,7 +2495,6 @@ Clean up key rotation threads resources */
 UNIV_INTERN
 void
 fil_crypt_threads_cleanup()
-/*=======================*/
 {
 	if (!fil_crypt_threads_inited) {
 		return;
@@ -2552,62 +2509,26 @@ fil_crypt_threads_cleanup()
 }
 
 /*********************************************************************
-Mark a space as closing */
-UNIV_INTERN
-void
-fil_space_crypt_mark_space_closing(
-/*===============================*/
-	ulint			space,		/*!< in: tablespace id */
-	fil_space_crypt_t*	crypt_data)	/*!< in: crypt_data or NULL */
-{
-	if (!fil_crypt_threads_inited) {
-		return;
-	}
-
-	mutex_enter(&fil_crypt_threads_mutex);
-
-	if (!crypt_data) {
-		crypt_data = fil_space_get_crypt_data(space);
-	}
-
-	if (crypt_data == NULL) {
-		mutex_exit(&fil_crypt_threads_mutex);
-		return;
-	}
-
-	mutex_enter(&crypt_data->mutex);
-	mutex_exit(&fil_crypt_threads_mutex);
-	crypt_data->closing = true;
-	mutex_exit(&crypt_data->mutex);
-}
-
-/*********************************************************************
-Wait for crypt threads to stop accessing space */
+Wait for crypt threads to stop accessing space
+@param[in]	space		Tablespace */
 UNIV_INTERN
 void
 fil_space_crypt_close_tablespace(
-/*=============================*/
-	ulint	space)	/*!< in: Space id */
+	const fil_space_t*	space)
 {
-	if (!srv_encrypt_tables) {
+	if (!srv_encrypt_tables || !space->crypt_data) {
 		return;
 	}
 
 	mutex_enter(&fil_crypt_threads_mutex);
 
-	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space);
-
-	if (crypt_data == NULL || crypt_data->is_closing(false)) {
-		mutex_exit(&fil_crypt_threads_mutex);
-		return;
-	}
+	fil_space_crypt_t* crypt_data = space->crypt_data;
 
-	uint start = time(0);
-	uint last = start;
+	time_t start = time(0);
+	time_t last = start;
 
 	mutex_enter(&crypt_data->mutex);
 	mutex_exit(&fil_crypt_threads_mutex);
-	crypt_data->closing = true;
 
 	uint cnt = crypt_data->rotate_state.active_threads;
 	bool flushing = crypt_data->rotate_state.flushing;
@@ -2617,20 +2538,22 @@ fil_space_crypt_close_tablespace(
 		/* release dict mutex so that scrub threads can release their
 		* table references */
 		dict_mutex_exit_for_mysql();
+
 		/* wakeup throttle (all) sleepers */
 		os_event_set(fil_crypt_throttle_sleep_event);
+
 		os_thread_sleep(20000);
 		dict_mutex_enter_for_mysql();
 		mutex_enter(&crypt_data->mutex);
 		cnt = crypt_data->rotate_state.active_threads;
 		flushing = crypt_data->rotate_state.flushing;
 
-		uint now = time(0);
+		time_t now = time(0);
 
 		if (now >= last + 30) {
 			ib_logf(IB_LOG_LEVEL_WARN,
-				"Waited %u seconds to drop space: %lu.",
-				now - start, space);
+				"Waited %ld seconds to drop space: %s(" ULINTPF ").",
+				now - start, space->name, space->id);
 			last = now;
 		}
 	}
@@ -2640,22 +2563,23 @@ fil_space_crypt_close_tablespace(
 
 /*********************************************************************
 Get crypt status for a space (used by information_schema)
-return 0 if crypt data present */
+@param[in]	space		Tablespace
+@param[out]	status		Crypt status */
 UNIV_INTERN
-int
+void
 fil_space_crypt_get_status(
-/*=======================*/
-	ulint				id,		/*!< in: space id */
-	struct fil_space_crypt_status_t* status)	/*!< out: status  */
+	const fil_space_t*			space,
+	struct fil_space_crypt_status_t*	status)
 {
-	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(id);
-
 	memset(status, 0, sizeof(*status));
 
+	ut_ad(space->n_pending_ops > 0);
+	fil_space_crypt_t* crypt_data = space->crypt_data;
+	status->space = space->id;
+
 	if (crypt_data != NULL) {
-		status->space = id;
-		status->scheme = crypt_data->type;
 		mutex_enter(&crypt_data->mutex);
+		status->scheme = crypt_data->type;
 		status->keyserver_requests = crypt_data->keyserver_requests;
 		status->min_key_version = crypt_data->min_key_version;
 		status->key_id = crypt_data->key_id;
@@ -2669,8 +2593,6 @@ fil_space_crypt_get_status(
 				crypt_data->rotate_state.next_offset;
 			status->rotate_max_page_number =
 				crypt_data->rotate_state.max_offset;
-		} else {
-			status->rotating = false;
 		}
 
 		mutex_exit(&crypt_data->mutex);
@@ -2678,25 +2600,17 @@ fil_space_crypt_get_status(
 		if (srv_encrypt_tables || crypt_data->min_key_version) {
 			status->current_key_version =
 				fil_crypt_get_latest_key_version(crypt_data);
-		} else {
-			status->current_key_version = 0;
-		}
-	} else {
-		if (srv_encrypt_tables) {
-			os_event_set(fil_crypt_threads_event);
 		}
 	}
-
-	return crypt_data == NULL ? 1 : 0;
 }
 
 /*********************************************************************
-Return crypt statistics */
+Return crypt statistics
+@param[out]	stat		Crypt statistics */
 UNIV_INTERN
 void
 fil_crypt_total_stat(
-/*=================*/
-	fil_crypt_stat_t *stat)	/*!< out: Crypt statistics */
+	fil_crypt_stat_t *stat)
 {
 	mutex_enter(&crypt_stat_mutex);
 	*stat = crypt_stat;
@@ -2705,21 +2619,24 @@ fil_crypt_total_stat(
 
 /*********************************************************************
 Get scrub status for a space (used by information_schema)
-return 0 if data found */
+
+@param[in]	space		Tablespace
+@param[out]	status		Scrub status */
 UNIV_INTERN
-int
+void
 fil_space_get_scrub_status(
-/*=======================*/
-	ulint id,					/*!< in: space id */
-	struct fil_space_scrub_status_t* status)	/*!< out: status  */
+	const fil_space_t*			space,
+	struct fil_space_scrub_status_t*	status)
 {
-	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(id);
-
 	memset(status, 0, sizeof(*status));
 
+	ut_ad(space->n_pending_ops > 0);
+	fil_space_crypt_t* crypt_data = space->crypt_data;
+
+	status->space = space->id;
+
 	if (crypt_data != NULL) {
-		status->space = id;
-		status->compressed = fil_space_get_zip_size(id) > 0;
+		status->compressed = fsp_flags_get_zip_size(space->flags) > 0;
 		mutex_enter(&crypt_data->mutex);
 		status->last_scrub_completed =
 			crypt_data->rotate_state.scrubbing.last_scrub_completed;
@@ -2734,12 +2651,8 @@ fil_space_get_scrub_status(
 				crypt_data->rotate_state.next_offset;
 			status->current_scrub_max_page_number =
 				crypt_data->rotate_state.max_offset;
-		} else {
-			status->scrubbing = false;
 		}
 
 		mutex_exit(&crypt_data->mutex);
 	}
-
-	return crypt_data == NULL ? 1 : 0;
 }
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 09322f074d1..74a33b4e4ff 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -156,7 +156,11 @@ UNIV_INTERN mysql_pfs_key_t	fil_space_latch_key;
 
 /** The tablespace memory cache. This variable is NULL before the module is
 initialized. */
-fil_system_t*	fil_system	= NULL;
+UNIV_INTERN fil_system_t*	fil_system	= NULL;
+
+/** At this age or older a space/page will be rotated */
+UNIV_INTERN extern uint srv_fil_crypt_rotate_key_age;
+UNIV_INTERN extern ib_mutex_t fil_crypt_threads_mutex;
 
 /** Determine if (i) is a user tablespace id or not. */
 # define fil_is_user_tablespace_id(i) ((i) > srv_undo_tablespaces_open)
@@ -613,6 +617,7 @@ fil_node_open_file(
 		node->handle = os_file_create_simple_no_error_handling(
 			innodb_file_data_key, node->name, OS_FILE_OPEN,
 			OS_FILE_READ_ONLY, &success, 0);
+
 		if (!success) {
 			/* The following call prints an error message */
 			os_file_get_last_error(true);
@@ -664,6 +669,16 @@ fil_node_open_file(
 		const ulint space_id = fsp_header_get_space_id(page);
 		ulint flags = fsp_header_get_flags(page);
 
+		/* 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);
+			node->space->crypt_data = fil_space_read_crypt_data(space_id, page, offset);
+			node->space->page_0_crypt_read = true;
+		}
+
 		ut_free(buf2);
 		os_file_close(node->handle);
 
@@ -1452,17 +1467,24 @@ fil_space_truncate_start(
 /*******************************************************************//**
 Creates a space memory object and puts it to the 'fil system' hash table.
 If there is an error, prints an error message to the .err log.
+@param[in]	name		Space name
+@param[in]	id		Space id
+@param[in]	flags		Tablespace flags
+@param[in]	purpose		FIL_TABLESPACE or FIL_LOG if log
+@param[in]	crypt_data	Encryption information
+@param[in]	create_table	True if this is create table
+@param[in]	mode		Encryption mode
 @return	TRUE if success */
 UNIV_INTERN
-ibool
+bool
 fil_space_create(
-/*=============*/
-	const char*	name,	/*!< in: space name */
-	ulint		id,	/*!< in: space id */
-	ulint		flags,	/*!< in: tablespace flags */
-	ulint		purpose,/*!< in: FIL_TABLESPACE, or FIL_LOG if log */
-	fil_space_crypt_t* crypt_data, /*!< in: crypt data */
-	bool		create_table) /*!< in: true if create table */
+	const char*		name,
+	ulint			id,
+	ulint			flags,
+	ulint			purpose,
+	fil_space_crypt_t*	crypt_data,
+	bool			create_table,
+	fil_encryption_t	mode)
 {
 	fil_space_t*	space;
 
@@ -1486,7 +1508,7 @@ fil_space_create(
 
 				mutex_exit(&fil_system->mutex);
 
-				return(FALSE);
+				return(false);
 			}
 
 			ib_logf(IB_LOG_LEVEL_WARN,
@@ -1513,7 +1535,7 @@ fil_space_create(
 
 		mutex_exit(&fil_system->mutex);
 
-		return(FALSE);
+		return(false);
 	}
 
 	space = static_cast<fil_space_t*>(mem_zalloc(sizeof(*space)));
@@ -1544,7 +1566,6 @@ fil_space_create(
 	space->flags = flags;
 
 	space->magic_n = FIL_SPACE_MAGIC_N;
-	space->printed_compression_failure = false;
 	space->crypt_data = crypt_data;
 
 	/* In create table we write page 0 so we have already
@@ -1569,14 +1590,28 @@ fil_space_create(
 
 	HASH_INSERT(fil_space_t, name_hash, fil_system->name_hash,
 		    ut_fold_string(name), space);
-	space->is_in_unflushed_spaces = false;
 
 	UT_LIST_ADD_LAST(space_list, fil_system->space_list, space);
 
 
-	mutex_exit(&fil_system->mutex);
+	/* Inform key rotation that there could be something
+	to do */
+	if (purpose == FIL_TABLESPACE && !srv_fil_crypt_rotate_key_age && fil_crypt_threads_event &&
+	    (mode == FIL_ENCRYPTION_ON || mode == FIL_ENCRYPTION_OFF ||
+		    srv_encrypt_tables)) {
+		/* Key rotation is not enabled, need to inform background
+		encryption threads. */
+		UT_LIST_ADD_LAST(rotation_list, fil_system->rotation_list, space);
+		mutex_exit(&fil_system->mutex);
+		space->is_in_rotation_list = true;
+		mutex_enter(&fil_crypt_threads_mutex);
+		os_event_set(fil_crypt_threads_event);
+		mutex_exit(&fil_crypt_threads_mutex);
+	} else {
+		mutex_exit(&fil_system->mutex);
+	}
 
-	return(TRUE);
+	return(true);
 }
 
 /*******************************************************************//**
@@ -1688,6 +1723,11 @@ fil_space_free(
 			       space);
 	}
 
+	if (space->is_in_rotation_list) {
+		space->is_in_rotation_list = false;
+		UT_LIST_REMOVE(rotation_list, fil_system->rotation_list, space);
+	}
+
 	UT_LIST_REMOVE(space_list, fil_system->space_list, space);
 
 	ut_a(space->magic_n == FIL_SPACE_MAGIC_N);
@@ -2344,7 +2384,6 @@ fil_read_first_page(
 	const char*	check_msg = NULL;
 	fil_space_crypt_t* cdata;
 
-
 	buf = static_cast<byte*>(ut_malloc(2 * UNIV_PAGE_SIZE));
 
 	/* Align the memory for a possible read from a raw device */
@@ -2383,7 +2422,7 @@ fil_read_first_page(
 
 	ulint space = fsp_header_get_space_id(page);
 	ulint offset = fsp_header_get_crypt_offset(
-		fsp_flags_get_zip_size(*flags), NULL);
+		fsp_flags_get_zip_size(*flags));
 
 	cdata = fil_space_read_crypt_data(space, page, offset);
 
@@ -2768,7 +2807,7 @@ fil_op_log_parse_or_replay(
 				    space_id, name, path, flags,
 				    DICT_TF2_USE_TABLESPACE,
 				    FIL_IBD_FILE_INITIAL_SIZE,
-				    FIL_SPACE_ENCRYPTION_DEFAULT,
+				    FIL_ENCRYPTION_DEFAULT,
 				    FIL_DEFAULT_ENCRYPTION_KEY) != DB_SUCCESS) {
 				ut_error;
 			}
@@ -2892,16 +2931,27 @@ fil_check_pending_operations(
 
 	*space = 0;
 
-	/* Wait for crypt threads to stop accessing space */
-	fil_space_crypt_close_tablespace(id);
-
 	mutex_enter(&fil_system->mutex);
 	fil_space_t* sp = fil_space_get_by_id(id);
+
 	if (sp) {
 		sp->stop_new_ops = TRUE;
+		/* space could be freed by other threads as soon
+		as n_pending_ops reaches 0, thus increment pending
+		ops here. */
+		sp->n_pending_ops++;
 	}
+
 	mutex_exit(&fil_system->mutex);
 
+	/* Wait for crypt threads to stop accessing space */
+	if (sp) {
+		fil_space_crypt_close_tablespace(sp);
+		/* We have "acquired" this space and must
+		free it now as below we compare n_pending_ops. */
+		fil_space_release(sp);
+	}
+
 	/* Check for pending change buffer merges. */
 
 	do {
@@ -3944,13 +3994,13 @@ fil_create_new_single_table_tablespace(
 
 	/* Create crypt data if the tablespace is either encrypted or user has
 	requested it to remain unencrypted. */
-	if (mode == FIL_SPACE_ENCRYPTION_ON || mode == FIL_SPACE_ENCRYPTION_OFF ||
+	if (mode == FIL_ENCRYPTION_ON || mode == FIL_ENCRYPTION_OFF ||
 		srv_encrypt_tables) {
 		crypt_data = fil_space_create_crypt_data(mode, key_id);
 	}
 
 	success = fil_space_create(tablename, space_id, flags, FIL_TABLESPACE,
-				   crypt_data, true);
+				   crypt_data, true, mode);
 
 	if (!success || !fil_node_create(path, size, space_id, FALSE)) {
 		err = DB_ERROR;
@@ -6633,7 +6683,8 @@ fil_iterate(
 				page_type == FIL_PAGE_PAGE_COMPRESSED);
 
 			/* If tablespace is encrypted, we need to decrypt
-			the page. */
+			the page. Note that tablespaces are not in
+			fil_system during import. */
 			if (encrypted) {
 				decrypted = fil_space_decrypt(
 							iter.crypt_data,
@@ -6886,8 +6937,11 @@ fil_tablespace_iterate(
 		iter.n_io_buffers = n_io_buffers;
 		iter.page_size = callback.get_page_size();
 
+		/* In MariaDB/MySQL 5.6 tablespace does not exist
+		during import, therefore we can't use space directly
+		here. */
 		ulint crypt_data_offset = fsp_header_get_crypt_offset(
-			callback.get_zip_size(), 0);
+			callback.get_zip_size());
 
 		/* read (optional) crypt data */
 		iter.crypt_data = fil_space_read_crypt_data(
@@ -6929,7 +6983,7 @@ fil_tablespace_iterate(
 
 		mem_free(io_buffer);
 
-		if (iter.crypt_data != NULL) {
+		if (crypt_io_buffer != NULL) {
 			mem_free(crypt_io_buffer);
 			iter.crypt_io_buffer = NULL;
 			fil_space_destroy_crypt_data(&iter.crypt_data);
@@ -7176,36 +7230,6 @@ fil_get_first_space()
 }
 
 /******************************************************************
-Get id of first tablespace that has node or ULINT_UNDEFINED if none */
-UNIV_INTERN
-ulint
-fil_get_first_space_safe()
-/*======================*/
-{
-	ulint out_id = ULINT_UNDEFINED;
-	fil_space_t* space;
-
-	mutex_enter(&fil_system->mutex);
-
-	space = UT_LIST_GET_FIRST(fil_system->space_list);
-	if (space != NULL) {
-		do
-		{
-			if (!space->stop_new_ops && UT_LIST_GET_LEN(space->chain) > 0) {
-				out_id = space->id;
-				break;
-			}
-
-			space = UT_LIST_GET_NEXT(space_list, space);
-		} while (space != NULL);
-	}
-
-	mutex_exit(&fil_system->mutex);
-
-	return out_id;
-}
-
-/******************************************************************
 Get id of next tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
@@ -7246,165 +7270,205 @@ fil_get_next_space(
 	return out_id;
 }
 
-/******************************************************************
-Get id of next tablespace that has node or ULINT_UNDEFINED if none */
-UNIV_INTERN
-ulint
-fil_get_next_space_safe(
-/*====================*/
-	ulint	id)	/*!< in: previous space id */
+/** Acquire a tablespace when it could be dropped concurrently.
+Used by background threads that do not necessarily hold proper locks
+for concurrency control.
+@param[in]	id	tablespace ID
+@param[in]	silent	whether to silently ignore missing tablespaces
+@return the tablespace, or NULL if missing or being deleted */
+inline
+fil_space_t*
+fil_space_acquire_low(
+	ulint	id,
+	bool	silent)
 {
-	bool found;
-	fil_space_t* space;
-	ulint out_id = ULINT_UNDEFINED;
+	fil_space_t*	space;
 
 	mutex_enter(&fil_system->mutex);
 
 	space = fil_space_get_by_id(id);
-	if (space == NULL) {
-		/* we didn't find it...search for space with space->id > id */
-		found = false;
-		space = UT_LIST_GET_FIRST(fil_system->space_list);
-	} else {
-		/* we found it, take next available space */
-		found = true;
-	}
-
-	while ((space = UT_LIST_GET_NEXT(space_list, space)) != NULL) {
 
-		if (!found && space->id <= id)
-			continue;
-
-		if (!space->stop_new_ops) {
-			/* inc reference to prevent drop */
-			out_id = space->id;
-			break;
+	if (space == NULL) {
+		if (!silent) {
+			ib_logf(IB_LOG_LEVEL_WARN, "Trying to access missing"
+				" tablespace " ULINTPF ".", id);
 		}
+	} else if (space->stop_new_ops) {
+		space = NULL;
+	} else {
+		space->n_pending_ops++;
 	}
 
 	mutex_exit(&fil_system->mutex);
 
-	return out_id;
+	return(space);
 }
 
-/******************************************************************
-Get crypt data for a tablespace */
-UNIV_INTERN
-fil_space_crypt_t*
-fil_space_get_crypt_data(
-/*=====================*/
-	ulint id)	/*!< in: space id */
+/** Acquire a tablespace when it could be dropped concurrently.
+Used by background threads that do not necessarily hold proper locks
+for concurrency control.
+@param[in]	id	tablespace ID
+@return the tablespace, or NULL if missing or being deleted */
+fil_space_t*
+fil_space_acquire(
+	ulint	id)
 {
-	fil_space_t*	space;
-	fil_space_crypt_t* crypt_data = NULL;
+	return(fil_space_acquire_low(id, false));
+}
 
-	ut_ad(fil_system);
+/** Acquire a tablespace that may not exist.
+Used by background threads that do not necessarily hold proper locks
+for concurrency control.
+@param[in]	id	tablespace ID
+@return the tablespace, or NULL if missing or being deleted */
+fil_space_t*
+fil_space_acquire_silent(
+	ulint	id)
+{
+	return(fil_space_acquire_low(id, true));
+}
 
+/** Release a tablespace acquired with fil_space_acquire().
+@param[in,out]	space	tablespace to release  */
+void
+fil_space_release(
+	fil_space_t*	space)
+{
 	mutex_enter(&fil_system->mutex);
+	ut_ad(space->magic_n == FIL_SPACE_MAGIC_N);
+	ut_ad(space->n_pending_ops > 0);
+	space->n_pending_ops--;
+	mutex_exit(&fil_system->mutex);
+}
 
-	space = fil_space_get_by_id(id);
+/** Return the next fil_space_t.
+Once started, the caller must keep calling this until it returns NULL.
+fil_space_acquire() and fil_space_release() are invoked here which
+blocks a concurrent operation from dropping the tablespace.
+@param[in]	prev_space	Pointer to the previous fil_space_t.
+If NULL, use the first fil_space_t on fil_system->space_list.
+@return pointer to the next fil_space_t.
+@retval NULL if this was the last*/
+fil_space_t*
+fil_space_next(
+	fil_space_t*	prev_space)
+{
+	fil_space_t*		space=prev_space;
 
-	mutex_exit(&fil_system->mutex);
+	mutex_enter(&fil_system->mutex);
 
-	if (space != NULL) {
-		/* If we have not yet read the page0
-		of this tablespace we will do it now. */
-		if (!space->crypt_data && !space->page_0_crypt_read) {
-			ulint space_id = space->id;
-			fil_node_t*	node;
-
-			ut_a(space->crypt_data == NULL);
-			node = UT_LIST_GET_FIRST(space->chain);
-
-			byte *buf = static_cast<byte*>(ut_malloc(2 * UNIV_PAGE_SIZE));
-			byte *page = static_cast<byte*>(ut_align(buf, UNIV_PAGE_SIZE));
-			fil_read(true, space_id, 0, 0, 0, UNIV_PAGE_SIZE, page,
-				NULL, NULL);
-			ulint offset = fsp_header_get_crypt_offset(
-				fsp_header_get_zip_size(page), NULL);
-			space->crypt_data = fil_space_read_crypt_data(space_id, page, offset);
-			ut_free(buf);
+	if (prev_space == NULL) {
+		space = UT_LIST_GET_FIRST(fil_system->space_list);
 
-#ifdef UNIV_DEBUG
-			ib_logf(IB_LOG_LEVEL_INFO,
-				"Read page 0 from tablespace for space %lu name %s key_id %u encryption %d handle %d.",
-				space_id,
-				space->name,
-				space->crypt_data ? space->crypt_data->key_id : 0,
-				space->crypt_data ? space->crypt_data->encryption : 0,
-				node->handle);
-#endif
+		/* We can trust that space is not NULL because at least the
+		system tablespace is always present and loaded first. */
+		space->n_pending_ops++;
+	} else {
+		ut_ad(space->n_pending_ops > 0);
 
-			ut_a(space->id == space_id);
+		/* Move on to the next fil_space_t */
+		space->n_pending_ops--;
+		space = UT_LIST_GET_NEXT(space_list, space);
 
-			space->page_0_crypt_read = true;
+		/* Skip spaces that are being created by
+		fil_ibd_create(), or dropped, or !tablespace. */
+		while (space != NULL
+			&& (UT_LIST_GET_LEN(space->chain) == 0
+				|| space->stop_new_ops
+				|| space->purpose != FIL_TABLESPACE)) {
+			space = UT_LIST_GET_NEXT(space_list, space);
 		}
 
-		crypt_data = space->crypt_data;
-
-		if (!space->page_0_crypt_read) {
-			ib_logf(IB_LOG_LEVEL_WARN,
-				"Space %lu name %s contains encryption %d information for key_id %u but page0 is not read.",
-				space->id,
-				space->name,
-				space->crypt_data ? space->crypt_data->encryption : 0,
-				space->crypt_data ? space->crypt_data->key_id : 0);
+		if (space != NULL) {
+			space->n_pending_ops++;
 		}
 	}
 
-	return(crypt_data);
+	mutex_exit(&fil_system->mutex);
+
+	return(space);
 }
 
-/******************************************************************
-Get crypt data for a tablespace */
-UNIV_INTERN
-fil_space_crypt_t*
-fil_space_set_crypt_data(
-/*=====================*/
-	ulint id, 	               /*!< in: space id */
-	fil_space_crypt_t* crypt_data) /*!< in: crypt data */
+/**
+Remove space from key rotation list if there are no more
+pending operations.
+@param[in]	space		Tablespace */
+static
+void
+fil_space_remove_from_keyrotation(
+	fil_space_t* space)
 {
-	fil_space_t*	space;
-	fil_space_crypt_t* free_crypt_data = NULL;
-	fil_space_crypt_t* ret_crypt_data = NULL;
+	ut_ad(mutex_own(&fil_system->mutex));
+	ut_ad(space);
 
-	ut_ad(fil_system);
+	if (space->n_pending_ops == 0) {
+		space->is_in_rotation_list = false;
+		UT_LIST_REMOVE(rotation_list, fil_system->rotation_list, space);
+	}
+}
 
-	mutex_enter(&fil_system->mutex);
 
-	space = fil_space_get_by_id(id);
+/** Return the next fil_space_t from key rotation list.
+Once started, the caller must keep calling this until it returns NULL.
+fil_space_acquire() and fil_space_release() are invoked here which
+blocks a concurrent operation from dropping the tablespace.
+@param[in]	prev_space	Pointer to the previous fil_space_t.
+If NULL, use the first fil_space_t on fil_system->space_list.
+@return pointer to the next fil_space_t.
+@retval NULL if this was the last*/
+fil_space_t*
+fil_space_keyrotate_next(
+	fil_space_t*	prev_space)
+{
+	fil_space_t* space = prev_space;
+	fil_space_t* old   = NULL;
 
-	if (space != NULL) {
-		if (space->crypt_data != NULL) {
-			/* Here we need to release fil_system mutex to
-			avoid mutex deadlock assertion. Here we would
-			taje mutexes in order fil_system, crypt_data and
-			in fil_crypt_start_encrypting_space we would
-			take them in order crypt_data, fil_system
-			at fil_space_get_flags -> fil_space_get_space */
-			mutex_exit(&fil_system->mutex);
-			fil_space_merge_crypt_data(space->crypt_data,
-						   crypt_data);
-			ret_crypt_data = space->crypt_data;
-			free_crypt_data = crypt_data;
-		} else {
-			space->crypt_data = crypt_data;
-			ret_crypt_data = space->crypt_data;
-			mutex_exit(&fil_system->mutex);
+	mutex_enter(&fil_system->mutex);
+
+	if (UT_LIST_GET_LEN(fil_system->rotation_list) == 0) {
+		if (space) {
+			ut_ad(space->n_pending_ops > 0);
+			space->n_pending_ops--;
+			fil_space_remove_from_keyrotation(space);
 		}
-	} else {
-		/* there is a small risk that tablespace has been deleted */
-		free_crypt_data = crypt_data;
 		mutex_exit(&fil_system->mutex);
+		return(NULL);
+	}
+
+	if (prev_space == NULL) {
+		space = UT_LIST_GET_FIRST(fil_system->rotation_list);
+
+		/* We can trust that space is not NULL because we
+		checked list length above */
+	} else {
+		ut_ad(space->n_pending_ops > 0);
+
+		/* Move on to the next fil_space_t */
+		space->n_pending_ops--;
+
+		old = space;
+		space = UT_LIST_GET_NEXT(rotation_list, space);
+
+		fil_space_remove_from_keyrotation(old);
+	}
+
+	/* Skip spaces that are being created by fil_ibd_create(),
+	or dropped. Note that rotation_list contains only
+	space->purpose == FIL_TABLESPACE. */
+	while (space != NULL
+		&& (UT_LIST_GET_LEN(space->chain) == 0
+			|| space->stop_new_ops)) {
+
+		old = space;
+		space = UT_LIST_GET_NEXT(rotation_list, space);
+		fil_space_remove_from_keyrotation(old);
 	}
 
-	if (free_crypt_data != NULL) {
-		/* there was already crypt data present and the new crypt
-		* data provided as argument to this function has been merged
-		* into that => free new crypt data
-		*/
-		fil_space_destroy_crypt_data(&free_crypt_data);
+	if (space != NULL) {
+		space->n_pending_ops++;
 	}
 
-	return ret_crypt_data;
+	mutex_exit(&fil_system->mutex);
+
+	return(space);
 }
diff --git a/storage/innobase/fsp/fsp0fsp.cc b/storage/innobase/fsp/fsp0fsp.cc
index c5289b48ccc..a40283412e2 100644
--- a/storage/innobase/fsp/fsp0fsp.cc
+++ b/storage/innobase/fsp/fsp0fsp.cc
@@ -677,7 +677,7 @@ UNIV_INTERN
 void
 fsp_header_init(
 /*============*/
-	ulint	space,		/*!< in: space id */
+	ulint	space_id,	/*!< in: space id */
 	ulint	size,		/*!< in: current size in blocks */
 	mtr_t*	mtr)		/*!< in/out: mini-transaction */
 {
@@ -689,11 +689,11 @@ fsp_header_init(
 
 	ut_ad(mtr);
 
-	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	mtr_x_lock(fil_space_get_latch(space_id, &flags), mtr);
 
 	zip_size = fsp_flags_get_zip_size(flags);
-	block = buf_page_create(space, 0, zip_size, mtr);
-	buf_page_get(space, zip_size, 0, RW_X_LATCH, mtr);
+	block = buf_page_create(space_id, 0, zip_size, mtr);
+	buf_page_get(space_id, zip_size, 0, RW_X_LATCH, mtr);
 	buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 
 	/* The prior contents of the file page should be ignored */
@@ -706,7 +706,7 @@ fsp_header_init(
 
 	header = FSP_HEADER_OFFSET + page;
 
-	mlog_write_ulint(header + FSP_SPACE_ID, space, MLOG_4BYTES, mtr);
+	mlog_write_ulint(header + FSP_SPACE_ID, space_id, MLOG_4BYTES, mtr);
 	mlog_write_ulint(header + FSP_NOT_USED, 0, MLOG_4BYTES, mtr);
 
 	mlog_write_ulint(header + FSP_SIZE, size, MLOG_4BYTES, mtr);
@@ -722,18 +722,23 @@ fsp_header_init(
 	flst_init(header + FSP_SEG_INODES_FREE, mtr);
 
 	mlog_write_ull(header + FSP_SEG_ID, 1, mtr);
-	if (space == 0) {
-		fsp_fill_free_list(FALSE, space, header, mtr);
+	if (space_id == 0) {
+		fsp_fill_free_list(FALSE, space_id, header, mtr);
 		btr_create(DICT_CLUSTERED | DICT_UNIVERSAL | DICT_IBUF,
-			   0, 0, DICT_IBUF_ID_MIN + space,
+			   0, 0, DICT_IBUF_ID_MIN + space_id,
 			   dict_ind_redundant, mtr);
 	} else {
-		fsp_fill_free_list(TRUE, space, header, mtr);
+		fsp_fill_free_list(TRUE, space_id, header, mtr);
 	}
 
-	ulint maxsize = 0;
-	ulint offset = fsp_header_get_crypt_offset(zip_size, &maxsize);
-	fil_space_write_crypt_data(space, page, offset, maxsize, mtr);
+	fil_space_t* space = fil_space_acquire(space_id);
+	ut_ad(space);
+
+	if (space->crypt_data) {
+		space->crypt_data->write_page0(page, mtr);
+	}
+
+	fil_space_release(space);
 }
 
 #endif /* !UNIV_HOTBACKUP */
@@ -4123,12 +4128,11 @@ fsp_print(
 
 /**********************************************************************//**
 Compute offset after xdes where crypt data can be stored
+@param[in]	zip_size	Compressed size or 0
 @return	offset */
 ulint
 fsp_header_get_crypt_offset(
-/*========================*/
-	ulint   zip_size, /*!< in: zip_size */
-	ulint*  max_size) /*!< out: free space available for crypt data */
+	const ulint   zip_size)
 {
 	ulint pageno = 0;
 	/* compute first page_no that will have xdes stored on page != 0*/
@@ -4143,12 +4147,6 @@ fsp_header_get_crypt_offset(
 	ulint iv_offset = XDES_ARR_OFFSET +
 		XDES_SIZE * (1 + xdes_calc_descriptor_index(zip_size, pageno));
 
-	if (max_size != NULL) {
-		/* return how much free space there is available on page */
-		*max_size = (zip_size ? zip_size : UNIV_PAGE_SIZE) -
-			(FSP_HEADER_OFFSET + iv_offset + FIL_PAGE_DATA_END);
-	}
-
 	return FSP_HEADER_OFFSET + iv_offset;
 }
 
diff --git a/storage/innobase/fts/fts0fts.cc b/storage/innobase/fts/fts0fts.cc
index 124837e5151..bfe32e2aec4 100644
--- a/storage/innobase/fts/fts0fts.cc
+++ b/storage/innobase/fts/fts0fts.cc
@@ -1989,7 +1989,8 @@ fts_create_one_index_table(
 	dict_mem_table_add_col(new_table, heap, "ilist", DATA_BLOB,
 			       4130048,	0);
 
-	error = row_create_table_for_mysql(new_table, trx, false, FIL_SPACE_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+	error = row_create_table_for_mysql(new_table, trx, false,
+		FIL_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
 
 	if (error != DB_SUCCESS) {
 		trx->error_state = error;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 54ae4c5dcf5..c9d78d68092 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -1084,6 +1084,9 @@ static SHOW_VAR innodb_status_variables[]= {
   {"encryption_rotation_estimated_iops",
   (char*) &export_vars.innodb_encryption_rotation_estimated_iops,
    SHOW_LONG},
+  {"encryption_key_rotation_list_length",
+  (char*)&export_vars.innodb_key_rotation_list_length,
+   SHOW_LONGLONG},
 
   /* scrubing */
   {"scrub_background_page_reorganizations",
@@ -11752,7 +11755,7 @@ ha_innobase::check_table_options(
 	atomic_writes_t awrites = (atomic_writes_t)options->atomic_writes;
 	fil_encryption_t encrypt = (fil_encryption_t)options->encryption;
 
-	if (encrypt != FIL_SPACE_ENCRYPTION_DEFAULT && !use_tablespace) {
+	if (encrypt != FIL_ENCRYPTION_DEFAULT && !use_tablespace) {
 		push_warning(
 			thd, Sql_condition::WARN_LEVEL_WARN,
 			HA_WRONG_CREATE_OPTION,
@@ -11760,7 +11763,7 @@ ha_innobase::check_table_options(
 		return "ENCRYPTED";
  	}
 
-        if (encrypt == FIL_SPACE_ENCRYPTION_OFF && srv_encrypt_tables == 2) {
+        if (encrypt == FIL_ENCRYPTION_OFF && srv_encrypt_tables == 2) {
 		push_warning(
 			thd, Sql_condition::WARN_LEVEL_WARN,
 			HA_WRONG_CREATE_OPTION,
@@ -11841,8 +11844,8 @@ ha_innobase::check_table_options(
 	}
 
 	/* If encryption is set up make sure that used key_id is found */
-	if (encrypt == FIL_SPACE_ENCRYPTION_ON ||
-            (encrypt == FIL_SPACE_ENCRYPTION_DEFAULT && srv_encrypt_tables)) {
+	if (encrypt == FIL_ENCRYPTION_ON ||
+            (encrypt == FIL_ENCRYPTION_DEFAULT && srv_encrypt_tables)) {
 		if (!encryption_key_id_exists((unsigned int)options->encryption_key_id)) {
 			push_warning_printf(
 				thd, Sql_condition::WARN_LEVEL_WARN,
@@ -11855,7 +11858,7 @@ ha_innobase::check_table_options(
 	}
 
 	/* Ignore nondefault key_id if encryption is set off */
-	if (encrypt == FIL_SPACE_ENCRYPTION_OFF &&
+	if (encrypt == FIL_ENCRYPTION_OFF &&
 		options->encryption_key_id != THDVAR(thd, default_encryption_key_id)) {
 		push_warning_printf(
 			thd, Sql_condition::WARN_LEVEL_WARN,
@@ -11868,7 +11871,7 @@ ha_innobase::check_table_options(
 
 	/* If default encryption is used make sure that used kay is found
 	from key file. */
-	if (encrypt == FIL_SPACE_ENCRYPTION_DEFAULT &&
+	if (encrypt == FIL_ENCRYPTION_DEFAULT &&
 		!srv_encrypt_tables &&
 		options->encryption_key_id != FIL_DEFAULT_ENCRYPTION_KEY) {
 		if (!encryption_key_id_exists((unsigned int)options->encryption_key_id)) {
@@ -19820,10 +19823,11 @@ static MYSQL_SYSVAR_UINT(encryption_rotate_key_age,
 			 PLUGIN_VAR_RQCMDARG,
 			 "Key rotation - re-encrypt in background "
                          "all pages that were encrypted with a key that "
-                         "many (or more) versions behind",
+                         "many (or more) versions behind. Value 0 indicates "
+			 "that key rotation is disabled.",
 			 NULL,
 			 innodb_encryption_rotate_key_age_update,
-			 srv_fil_crypt_rotate_key_age, 0, UINT_MAX32, 0);
+			 1, 0, UINT_MAX32, 0);
 
 static MYSQL_SYSVAR_UINT(encryption_rotation_iops, srv_n_fil_crypt_iops,
 			 PLUGIN_VAR_RQCMDARG,
@@ -20737,8 +20741,9 @@ innodb_encrypt_tables_validate(
 						for update function */
 	struct st_mysql_value*		value)	/*!< in: incoming string */
 {
-	if (check_sysvar_enum(thd, var, save, value))
+	if (check_sysvar_enum(thd, var, save, value)) {
 		return 1;
+	}
 
 	ulong encrypt_tables = *(ulong*)save;
 
@@ -20750,6 +20755,17 @@ innodb_encrypt_tables_validate(
 		                    "encryption plugin is not available");
 		return 1;
 	}
+
+	if (!srv_fil_crypt_rotate_key_age) {
+		const char *msg = (encrypt_tables ? "enable" : "disable");
+		push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+				    HA_ERR_UNSUPPORTED,
+				    "InnoDB: cannot %s encryption, "
+				    "innodb_encryption_rotate_key_age=0"
+				    " i.e. key rotation disabled", msg);
+		return 1;
+	}
+
 	return 0;
 }
 
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index b12e5223a34..0af8c765156 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -2883,9 +2883,11 @@ prepare_inplace_alter_table_dict(
 		ulint		n_cols;
 		dtuple_t*	add_cols;
 		ulint		key_id = FIL_DEFAULT_ENCRYPTION_KEY;
-		fil_encryption_t mode = FIL_SPACE_ENCRYPTION_DEFAULT;
+		fil_encryption_t mode = FIL_ENCRYPTION_DEFAULT;
 
-		crypt_data = fil_space_get_crypt_data(ctx->prebuilt->table->space);
+		fil_space_t* space = fil_space_acquire(ctx->prebuilt->table->space);
+		crypt_data = space->crypt_data;
+		fil_space_release(space);
 
 		if (crypt_data) {
 			key_id = crypt_data->key_id;
diff --git a/storage/innobase/handler/i_s.cc b/storage/innobase/handler/i_s.cc
index a73446440aa..c856d03864f 100644
--- a/storage/innobase/handler/i_s.cc
+++ b/storage/innobase/handler/i_s.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2007, 2016, Oracle and/or its affiliates.
-Copyrigth (c) 2014, 2016, MariaDB Corporation
+Copyrigth (c) 2014, 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
@@ -8180,22 +8180,31 @@ static ST_FIELD_INFO	innodb_tablespaces_encryption_fields_info[] =
 	 STRUCT_FLD(old_name,		""),
 	 STRUCT_FLD(open_method,	SKIP_OPEN_TABLE)},
 
+#define TABLESPACES_ENCRYPTION_ROTATING_OR_FLUSHING 9
+	{STRUCT_FLD(field_name,		"ROTATING_OR_FLUSHING"),
+	 STRUCT_FLD(field_length,	1),
+	 STRUCT_FLD(field_type,		MYSQL_TYPE_LONG),
+	 STRUCT_FLD(value,		0),
+	 STRUCT_FLD(field_flags,	MY_I_S_UNSIGNED),
+	 STRUCT_FLD(old_name,		""),
+	 STRUCT_FLD(open_method,	SKIP_OPEN_TABLE)},
+
 	END_OF_ST_FIELD_INFO
 };
 
 /**********************************************************************//**
 Function to fill INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION
-with information collected by scanning SYS_TABLESPACES table and then use
-fil_space()
+with information collected by scanning SYS_TABLESPACES table.
+@param[in]	thd		thread handle
+@param[in]	space		Tablespace
+@param[in]	table_to_fill	I_S table to fill
 @return 0 on success */
 static
 int
 i_s_dict_fill_tablespaces_encryption(
-/*==========================*/
-	THD*		thd,		/*!< in: thread */
-	ulint		space,		/*!< in: space ID */
-	const char*	name,		/*!< in: tablespace name */
-	TABLE*		table_to_fill)	/*!< in/out: fill this table */
+	THD*		thd,
+	fil_space_t*	space,
+	TABLE*		table_to_fill)
 {
 	Field**	fields;
 	struct fil_space_crypt_status_t status;
@@ -8205,10 +8214,11 @@ i_s_dict_fill_tablespaces_encryption(
 	fields = table_to_fill->field;
 
 	fil_space_crypt_get_status(space, &status);
-	OK(fields[TABLESPACES_ENCRYPTION_SPACE]->store(space));
+
+	OK(fields[TABLESPACES_ENCRYPTION_SPACE]->store(space->id));
 
 	OK(field_store_string(fields[TABLESPACES_ENCRYPTION_NAME],
-			      name));
+			      space->name));
 
 	OK(fields[TABLESPACES_ENCRYPTION_ENCRYPTION_SCHEME]->store(
 		   status.scheme));
@@ -8220,6 +8230,9 @@ i_s_dict_fill_tablespaces_encryption(
 		   status.current_key_version));
 	OK(fields[TABLESPACES_ENCRYPTION_CURRENT_KEY_ID]->store(
 		   status.key_id));
+	OK(fields[TABLESPACES_ENCRYPTION_ROTATING_OR_FLUSHING]->store(
+			(status.rotating || status.flushing) ? 1 : 0));
+
 	if (status.rotating) {
 		fields[TABLESPACES_ENCRYPTION_KEY_ROTATION_PAGE_NUMBER]->set_notnull();
 		OK(fields[TABLESPACES_ENCRYPTION_KEY_ROTATION_PAGE_NUMBER]->store(
@@ -8233,6 +8246,7 @@ i_s_dict_fill_tablespaces_encryption(
 		fields[TABLESPACES_ENCRYPTION_KEY_ROTATION_MAX_PAGE_NUMBER]
 			->set_null();
 	}
+
 	OK(schema_table_store_record(thd, table_to_fill));
 
 	DBUG_RETURN(0);
@@ -8272,30 +8286,36 @@ i_s_tablespaces_encryption_fill_table(
 
 	while (rec) {
 		const char*	err_msg;
-		ulint		space;
+		ulint		space_id;
 		const char*	name;
 		ulint		flags;
 
 		/* Extract necessary information from a SYS_TABLESPACES row */
 		err_msg = dict_process_sys_tablespaces(
-			heap, rec, &space, &name, &flags);
+			heap, rec, &space_id, &name, &flags);
 
 		mtr_commit(&mtr);
 		mutex_exit(&dict_sys->mutex);
 
-		if (space == 0) {
+		if (space_id == 0) {
 			found_space_0 = true;
 		}
 
-		if (!err_msg) {
+		fil_space_t* space = fil_space_acquire_silent(space_id);
+
+		if (!err_msg && space) {
 			i_s_dict_fill_tablespaces_encryption(
-				thd, space, name, tables->table);
+				thd, space, tables->table);
 		} else {
 			push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
 					    ER_CANT_FIND_SYSTEM_REC, "%s",
 					    err_msg);
 		}
 
+		if (space) {
+			fil_space_release(space);
+		}
+
 		mem_heap_empty(heap);
 
 		/* Get the next record */
@@ -8311,10 +8331,13 @@ i_s_tablespaces_encryption_fill_table(
 	if (found_space_0 == false) {
 		/* space 0 does for what ever unknown reason not show up
 		* in iteration above, add it manually */
-		ulint		space = 0;
-		const char*	name = NULL;
+
+		fil_space_t* space = fil_space_acquire_silent(0);
+
 		i_s_dict_fill_tablespaces_encryption(
-			thd, space, name, tables->table);
+			thd, space, tables->table);
+
+		fil_space_release(space);
 	}
 
 	DBUG_RETURN(0);
@@ -8465,22 +8488,32 @@ static ST_FIELD_INFO	innodb_tablespaces_scrubbing_fields_info[] =
 	 STRUCT_FLD(old_name,		""),
 	 STRUCT_FLD(open_method,	SKIP_OPEN_TABLE)},
 
+#define TABLESPACES_ENCRYPTION_ROTATING_OR_FLUSHING 9
+	{STRUCT_FLD(field_name,		"ROTATING_OR_FLUSHING"),
+	 STRUCT_FLD(field_length,	MY_INT32_NUM_DECIMAL_DIGITS),
+	 STRUCT_FLD(field_type,		MYSQL_TYPE_LONG),
+	 STRUCT_FLD(value,		0),
+	 STRUCT_FLD(field_flags,	MY_I_S_UNSIGNED),
+	 STRUCT_FLD(old_name,		""),
+	 STRUCT_FLD(open_method,	SKIP_OPEN_TABLE)},
+
 	END_OF_ST_FIELD_INFO
 };
 
 /**********************************************************************//**
 Function to fill INFORMATION_SCHEMA.INNODB_TABLESPACES_SCRUBBING
-with information collected by scanning SYS_TABLESPACES table and then use
-fil_space()
+with information collected by scanning SYS_TABLESPACES table and
+fil_space.
+@param[in]	thd		Thread handle
+@param[in]	space		Tablespace
+@param[in]	table_to_fill	I_S table
 @return 0 on success */
 static
 int
 i_s_dict_fill_tablespaces_scrubbing(
-/*==========================*/
-	THD*		thd,		/*!< in: thread */
-	ulint		space,		/*!< in: space ID */
-	const char*	name,		/*!< in: tablespace name */
-	TABLE*		table_to_fill)	/*!< in/out: fill this table */
+	THD*		thd,
+	fil_space_t*	space,
+	TABLE*		table_to_fill)
 {
 	Field**	fields;
         struct fil_space_scrub_status_t status;
@@ -8490,10 +8523,11 @@ i_s_dict_fill_tablespaces_scrubbing(
 	fields = table_to_fill->field;
 
 	fil_space_get_scrub_status(space, &status);
-	OK(fields[TABLESPACES_SCRUBBING_SPACE]->store(space));
+
+	OK(fields[TABLESPACES_SCRUBBING_SPACE]->store(space->id));
 
 	OK(field_store_string(fields[TABLESPACES_SCRUBBING_NAME],
-			      name));
+			      space->name));
 
 	OK(fields[TABLESPACES_SCRUBBING_COMPRESSED]->store(
 		   status.compressed ? 1 : 0));
@@ -8513,6 +8547,7 @@ i_s_dict_fill_tablespaces_scrubbing(
 		TABLESPACES_SCRUBBING_CURRENT_SCRUB_ACTIVE_THREADS,
 		TABLESPACES_SCRUBBING_CURRENT_SCRUB_PAGE_NUMBER,
 		TABLESPACES_SCRUBBING_CURRENT_SCRUB_MAX_PAGE_NUMBER };
+
 	if (status.scrubbing) {
 		for (uint i = 0; i < array_elements(field_numbers); i++) {
 			fields[field_numbers[i]]->set_notnull();
@@ -8532,6 +8567,7 @@ i_s_dict_fill_tablespaces_scrubbing(
 			fields[field_numbers[i]]->set_null();
 		}
 	}
+
 	OK(schema_table_store_record(thd, table_to_fill));
 
 	DBUG_RETURN(0);
@@ -8571,30 +8607,36 @@ i_s_tablespaces_scrubbing_fill_table(
 
 	while (rec) {
 		const char*	err_msg;
-		ulint		space;
+		ulint		space_id;
 		const char*	name;
 		ulint		flags;
 
 		/* Extract necessary information from a SYS_TABLESPACES row */
 		err_msg = dict_process_sys_tablespaces(
-			heap, rec, &space, &name, &flags);
+			heap, rec, &space_id, &name, &flags);
 
 		mtr_commit(&mtr);
 		mutex_exit(&dict_sys->mutex);
 
-		if (space == 0) {
+		if (space_id == 0) {
 			found_space_0 = true;
 		}
 
-		if (!err_msg) {
+		fil_space_t* space = fil_space_acquire_silent(space_id);
+
+		if (!err_msg && space) {
 			i_s_dict_fill_tablespaces_scrubbing(
-				thd, space, name, tables->table);
+				thd, space, tables->table);
 		} else {
 			push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
 					    ER_CANT_FIND_SYSTEM_REC, "%s",
 					    err_msg);
 		}
 
+		if (space) {
+			fil_space_release(space);
+		}
+
 		mem_heap_empty(heap);
 
 		/* Get the next record */
@@ -8610,10 +8652,12 @@ i_s_tablespaces_scrubbing_fill_table(
 	if (found_space_0 == false) {
 		/* space 0 does for what ever unknown reason not show up
 		* in iteration above, add it manually */
-		ulint		space = 0;
-		const char*	name = NULL;
+		fil_space_t* space = fil_space_acquire_silent(0);
+
 		i_s_dict_fill_tablespaces_scrubbing(
-			thd, space, name, tables->table);
+			thd, space, tables->table);
+
+		fil_space_release(space);
 	}
 
 	DBUG_RETURN(0);
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index 08f2967a495..1e4a258ca54 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, MariaDB Corporation.
+Copyright (c) 2016, 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
@@ -4554,7 +4554,7 @@ ibuf_merge_or_delete_for_page(
 	buf_block_t*	block,	/*!< in: if page has been read from
 				disk, pointer to the page x-latched,
 				else NULL */
-	ulint		space,	/*!< in: space id of the index page */
+	ulint		space_id,/*!< in: space id of the index page */
 	ulint		page_no,/*!< in: page number of the index page */
 	ulint		zip_size,/*!< in: compressed page size in bytes,
 				or 0 */
@@ -4571,21 +4571,21 @@ ibuf_merge_or_delete_for_page(
 	ulint		volume			= 0;
 #endif
 	page_zip_des_t*	page_zip		= NULL;
-	ibool		tablespace_being_deleted = FALSE;
 	ibool		corruption_noticed	= FALSE;
 	mtr_t		mtr;
+	fil_space_t*	space			= NULL;
 
 	/* Counts for merged & discarded operations. */
 	ulint		mops[IBUF_OP_COUNT];
 	ulint		dops[IBUF_OP_COUNT];
 
-	ut_ad(!block || buf_block_get_space(block) == space);
+	ut_ad(!block || buf_block_get_space(block) == space_id);
 	ut_ad(!block || buf_block_get_page_no(block) == page_no);
 	ut_ad(!block || buf_block_get_zip_size(block) == zip_size);
 	ut_ad(!block || buf_block_get_io_fix(block) == BUF_IO_READ);
 
 	if (srv_force_recovery >= SRV_FORCE_NO_IBUF_MERGE
-	    || trx_sys_hdr_page(space, page_no)) {
+	    || trx_sys_hdr_page(space_id, page_no)) {
 		return;
 	}
 
@@ -4599,7 +4599,7 @@ ibuf_merge_or_delete_for_page(
 	uncompressed page size always is a power-of-2 multiple of the
 	compressed page size. */
 
-	if (ibuf_fixed_addr_page(space, 0, page_no)
+	if (ibuf_fixed_addr_page(space_id, 0, page_no)
 	    || fsp_descr_page(0, page_no)) {
 		return;
 	}
@@ -4607,19 +4607,19 @@ ibuf_merge_or_delete_for_page(
 	if (UNIV_LIKELY(update_ibuf_bitmap)) {
 		ut_a(ut_is_2pow(zip_size));
 
-		if (ibuf_fixed_addr_page(space, zip_size, page_no)
+		if (ibuf_fixed_addr_page(space_id, zip_size, page_no)
 		    || fsp_descr_page(zip_size, page_no)) {
 			return;
 		}
 
-		/* If the following returns FALSE, we get the counter
+		/* If the following returns space, we get the counter
 		incremented, and must decrement it when we leave this
 		function. When the counter is > 0, that prevents tablespace
 		from being dropped. */
 
-		tablespace_being_deleted = fil_inc_pending_ops(space, true);
+		space = fil_space_acquire(space_id);
 
-		if (UNIV_UNLIKELY(tablespace_being_deleted)) {
+		if (UNIV_UNLIKELY(!space)) {
 			/* Do not try to read the bitmap page from space;
 			just delete the ibuf records for the page */
 
@@ -4632,7 +4632,7 @@ ibuf_merge_or_delete_for_page(
 			ibuf_mtr_start(&mtr);
 
 			bitmap_page = ibuf_bitmap_get_map_page(
-				space, page_no, zip_size, &mtr);
+				space_id, page_no, zip_size, &mtr);
 
 			if (bitmap_page &&
 			    fil_page_get_type(bitmap_page) != FIL_PAGE_TYPE_ALLOCATED) {
@@ -4646,15 +4646,15 @@ ibuf_merge_or_delete_for_page(
 			if (!bitmap_bits) {
 				/* No inserts buffered for this page */
 
-				if (!tablespace_being_deleted) {
-					fil_decr_pending_ops(space);
+				if (space) {
+					fil_space_release(space);
 				}
 
 				return;
 			}
 		}
 	} else if (block
-		   && (ibuf_fixed_addr_page(space, zip_size, page_no)
+		   && (ibuf_fixed_addr_page(space_id, zip_size, page_no)
 		      || fsp_descr_page(zip_size, page_no))) {
 
 		return;
@@ -4662,7 +4662,7 @@ ibuf_merge_or_delete_for_page(
 
 	heap = mem_heap_create(512);
 
-	search_tuple = ibuf_search_tuple_build(space, page_no, heap);
+	search_tuple = ibuf_search_tuple_build(space_id, page_no, heap);
 
 	if (block) {
 		/* Move the ownership of the x-latch on the page to this OS
@@ -4688,7 +4688,7 @@ ibuf_merge_or_delete_for_page(
 			fputs("  InnoDB: Dump of the ibuf bitmap page:\n",
 			      stderr);
 
-			bitmap_page = ibuf_bitmap_get_map_page(space, page_no,
+			bitmap_page = ibuf_bitmap_get_map_page(space_id, page_no,
 							       zip_size, &mtr);
 			if (bitmap_page == NULL)
 			{
@@ -4772,7 +4772,7 @@ loop:
 
 		/* Check if the entry is for this index page */
 		if (ibuf_rec_get_page_no(&mtr, rec) != page_no
-		    || ibuf_rec_get_space(&mtr, rec) != space) {
+		    || ibuf_rec_get_space(&mtr, rec) != space_id) {
 
 			if (block) {
 				page_header_reset_last_insert(
@@ -4839,7 +4839,7 @@ loop:
 				ut_ad(page_rec_is_user_rec(rec));
 				ut_ad(ibuf_rec_get_page_no(&mtr, rec)
 				      == page_no);
-				ut_ad(ibuf_rec_get_space(&mtr, rec) == space);
+				ut_ad(ibuf_rec_get_space(&mtr, rec) == space_id);
 
 				/* Mark the change buffer record processed,
 				so that it will not be merged again in case
@@ -4869,7 +4869,7 @@ loop:
 				buf_block_dbg_add_level(
 					block, SYNC_IBUF_TREE_NODE);
 
-				if (!ibuf_restore_pos(space, page_no,
+				if (!ibuf_restore_pos(space_id, page_no,
 						      search_tuple,
 						      BTR_MODIFY_LEAF,
 						      &pcur, &mtr)) {
@@ -4893,7 +4893,7 @@ loop:
 		}
 
 		/* Delete the record from ibuf */
-		if (ibuf_delete_rec(space, page_no, &pcur, search_tuple,
+		if (ibuf_delete_rec(space_id, page_no, &pcur, search_tuple,
 				    &mtr)) {
 			/* Deletion was pessimistic and mtr was committed:
 			we start from the beginning again */
@@ -4913,7 +4913,7 @@ reset_bit:
 		page_t*	bitmap_page;
 
 		bitmap_page = ibuf_bitmap_get_map_page(
-			space, page_no, zip_size, &mtr);
+			space_id, page_no, zip_size, &mtr);
 
 		ibuf_bitmap_page_set_bits(
 			bitmap_page, page_no, zip_size,
@@ -4954,9 +4954,8 @@ reset_bit:
 	mutex_exit(&ibuf_mutex);
 #endif /* HAVE_ATOMIC_BUILTINS */
 
-	if (update_ibuf_bitmap && !tablespace_being_deleted) {
-
-		fil_decr_pending_ops(space);
+	if (space) {
+		fil_space_release(space);
 	}
 
 #ifdef UNIV_IBUF_COUNT_DEBUG
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 6aae71ba12d..cfc2d850883 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -39,14 +39,6 @@ static const unsigned char CRYPT_MAGIC[MAGIC_SZ] = {
 /* This key will be used if nothing else is given */
 #define FIL_DEFAULT_ENCRYPTION_KEY ENCRYPTION_KEY_SYSTEM_DATA
 
-/** Enum values for encryption table option */
-typedef enum {
-	FIL_SPACE_ENCRYPTION_DEFAULT = 0,	/* Tablespace encrypted if
-						srv_encrypt_tables = ON */
-	FIL_SPACE_ENCRYPTION_ON = 1,		/* Tablespace is encrypted always */
-	FIL_SPACE_ENCRYPTION_OFF = 2		/* Tablespace is not encrypted */
-} fil_encryption_t;
-
 extern os_event_t fil_crypt_threads_event;
 
 /**
@@ -110,23 +102,21 @@ struct fil_space_rotate_state_t
 	} scrubbing;
 };
 
-struct fil_space_crypt_struct : st_encryption_scheme
+struct fil_space_crypt_t : st_encryption_scheme
 {
  public:
 	/** Constructor. Does not initialize the members!
 	The object is expected to be placed in a buffer that
 	has been zero-initialized. */
-	fil_space_crypt_struct(
+	fil_space_crypt_t(
 		ulint new_type,
 		uint new_min_key_version,
 		uint new_key_id,
-		ulint offset,
 		fil_encryption_t new_encryption)
 		: st_encryption_scheme(),
 		min_key_version(new_min_key_version),
-		page0_offset(offset),
+		page0_offset(0),
 		encryption(new_encryption),
-		closing(false),
 		key_found(),
 		rotate_state()
 	{
@@ -138,9 +128,9 @@ struct fil_space_crypt_struct : st_encryption_scheme
 		locker = crypt_data_scheme_locker;
 		type = new_type;
 
-		if (new_encryption == FIL_SPACE_ENCRYPTION_OFF ||
+		if (new_encryption == FIL_ENCRYPTION_OFF ||
 			(!srv_encrypt_tables &&
-			 new_encryption == FIL_SPACE_ENCRYPTION_DEFAULT)) {
+			 new_encryption == FIL_ENCRYPTION_DEFAULT)) {
 			type = CRYPT_SCHEME_UNENCRYPTED;
 		} else {
 			type = CRYPT_SCHEME_1;
@@ -149,9 +139,8 @@ struct fil_space_crypt_struct : st_encryption_scheme
 	}
 
 	/** Destructor */
-	~fil_space_crypt_struct()
+	~fil_space_crypt_t()
 	{
-		closing = true;
 		mutex_free(&mutex);
 	}
 
@@ -169,45 +158,36 @@ struct fil_space_crypt_struct : st_encryption_scheme
 
 	/** Returns true if tablespace should be encrypted */
 	bool should_encrypt() const {
-		return ((encryption == FIL_SPACE_ENCRYPTION_ON) ||
+		return ((encryption == FIL_ENCRYPTION_ON) ||
 			(srv_encrypt_tables &&
-				encryption == FIL_SPACE_ENCRYPTION_DEFAULT));
+				encryption == FIL_ENCRYPTION_DEFAULT));
 	}
 
 	/** Return true if tablespace is encrypted. */
 	bool is_encrypted() const {
-		return (encryption != FIL_SPACE_ENCRYPTION_OFF);
+		return (encryption != FIL_ENCRYPTION_OFF);
 	}
 
 	/** Return true if default tablespace encryption is used, */
 	bool is_default_encryption() const {
-		return (encryption == FIL_SPACE_ENCRYPTION_DEFAULT);
+		return (encryption == FIL_ENCRYPTION_DEFAULT);
 	}
 
 	/** Return true if tablespace is not encrypted. */
 	bool not_encrypted() const {
-		return (encryption == FIL_SPACE_ENCRYPTION_OFF);
+		return (encryption == FIL_ENCRYPTION_OFF);
 	}
 
-	/** Is this tablespace closing. */
-	bool is_closing(bool is_fixed) {
-		bool closed;
-		if (!is_fixed) {
-			mutex_enter(&mutex);
-		}
-		closed = closing;
-		if (!is_fixed) {
-			mutex_exit(&mutex);
-		}
-		return closed;
-	}
+	/** Write crypt data to a page (0)
+	@param[in,out]	page0		Page 0 where to write
+	@param[in,out]	mtr		Minitransaction */
+	void write_page0(byte* page0, mtr_t* mtr);
 
 	uint min_key_version; // min key version for this space
 	ulint page0_offset;   // byte offset on page 0 for crypt data
 	fil_encryption_t encryption; // Encryption setup
 
 	ib_mutex_t mutex;   // mutex protecting following variables
-	bool closing;	    // is tablespace being closed
 
 	/** Return code from encryption_key_get_latest_version.
         If ENCRYPTION_KEY_VERSION_INVALID encryption plugin
@@ -219,146 +199,175 @@ struct fil_space_crypt_struct : st_encryption_scheme
 	fil_space_rotate_state_t rotate_state;
 };
 
-/* structure containing encryption specification */
-typedef struct fil_space_crypt_struct fil_space_crypt_t;
+/** Status info about encryption */
+struct fil_space_crypt_status_t {
+	ulint space;             /*!< tablespace id */
+	ulint scheme;            /*!< encryption scheme */
+	uint  min_key_version;   /*!< min key version */
+	uint  current_key_version;/*!< current key version */
+	uint  keyserver_requests;/*!< no of key requests to key server */
+	ulint key_id;            /*!< current key_id */
+	bool rotating;           /*!< is key rotation ongoing */
+	bool flushing;           /*!< is flush at end of rotation ongoing */
+	ulint rotate_next_page_number; /*!< next page if key rotating */
+	ulint rotate_max_page_number;  /*!< max page if key rotating */
+};
+
+/** Statistics about encryption key rotation */
+struct fil_crypt_stat_t {
+	ulint pages_read_from_cache;
+	ulint pages_read_from_disk;
+	ulint pages_modified;
+	ulint pages_flushed;
+	ulint estimated_iops;
+};
+
+/** Status info about scrubbing */
+struct fil_space_scrub_status_t {
+	ulint space;             /*!< tablespace id */
+	bool compressed;        /*!< is space compressed  */
+	time_t last_scrub_completed;  /*!< when was last scrub completed */
+	bool scrubbing;               /*!< is scrubbing ongoing */
+	time_t current_scrub_started; /*!< when started current scrubbing */
+	ulint current_scrub_active_threads; /*!< current scrub active threads */
+	ulint current_scrub_page_number; /*!< current scrub page no */
+	ulint current_scrub_max_page_number; /*!< current scrub max page no */
+};
 
 /*********************************************************************
-Init global resources needed for tablespace encryption/decryption */
+Init space crypt */
 UNIV_INTERN
 void
 fil_space_crypt_init();
 
 /*********************************************************************
-Cleanup global resources needed for tablespace encryption/decryption */
+Cleanup space crypt */
 UNIV_INTERN
 void
 fil_space_crypt_cleanup();
 
-/*********************************************************************
-Create crypt data, i.e data that is used for a single tablespace */
-UNIV_INTERN
-fil_space_crypt_t *
-fil_space_create_crypt_data(
-/*========================*/
-	fil_encryption_t	encrypt_mode,	/*!< in: encryption mode */
-	uint			key_id);	/*!< in: encryption key id */
-
-/*********************************************************************
-Destroy crypt data */
-UNIV_INTERN
-void
-fil_space_destroy_crypt_data(
-/*=========================*/
-	fil_space_crypt_t **crypt_data); /*!< in/out: crypt data */
-
-/*********************************************************************
-Get crypt data for a space*/
-UNIV_INTERN
-fil_space_crypt_t *
-fil_space_get_crypt_data(
-/*=====================*/
-	ulint space); /*!< in: tablespace id */
+/******************************************************************
+Create a fil_space_crypt_t object
+@param[in]	encrypt_mode	FIL_ENCRYPTION_DEFAULT or
+				FIL_ENCRYPTION_ON or
+				FIL_ENCRYPTION_OFF
 
-/*********************************************************************
-Set crypt data for a space*/
+@param[in]	key_id		Encryption key id
+@return crypt object */
 UNIV_INTERN
 fil_space_crypt_t*
-fil_space_set_crypt_data(
-/*=====================*/
-	ulint space,                    /*!< in: tablespace id */
-	fil_space_crypt_t* crypt_data); /*!< in: crypt data to set */
+fil_space_create_crypt_data(
+	fil_encryption_t	encrypt_mode,
+	uint			key_id)
+	MY_ATTRIBUTE((warn_unused_result));
 
-/*********************************************************************
-Merge crypt data */
+/******************************************************************
+Merge fil_space_crypt_t object
+@param[in,out]	dst		Destination cryp data
+@param[in]	src		Source crypt data */
 UNIV_INTERN
 void
 fil_space_merge_crypt_data(
-/*=======================*/
-	fil_space_crypt_t* dst_crypt_data,  /*!< in: crypt_data */
-	const fil_space_crypt_t* src_crypt_data); /*!< in: crypt data */
+	fil_space_crypt_t* dst,
+	const fil_space_crypt_t* src);
 
-/*********************************************************************
-Read crypt data from buffer page */
+/******************************************************************
+Read crypt data from a page (0)
+@param[in]	space		space_id
+@param[in]	page		Page 0
+@param[in]	offset		Offset to crypt data
+@return crypt data from page 0 or NULL. */
 UNIV_INTERN
-fil_space_crypt_t *
+fil_space_crypt_t*
 fil_space_read_crypt_data(
-/*======================*/
-	ulint space,      /*!< in: tablespace id */
-	const byte* page, /*!< in: buffer page */
-	ulint offset);    /*!< in: offset where crypt data is stored */
+	ulint		space,
+	const byte*	page,
+	ulint		offset)
+	MY_ATTRIBUTE((warn_unused_result));
 
-/*********************************************************************
-Write crypt data to buffer page */
+/******************************************************************
+Free a crypt data object
+@param[in,out] crypt_data	crypt data to be freed */
 UNIV_INTERN
 void
-fil_space_write_crypt_data(
-/*=======================*/
-	ulint space,   /*!< in: tablespace id */
-	byte* page,    /*!< in: buffer page */
-	ulint offset,  /*!< in: offset where to store data */
-	ulint maxsize, /*!< in: max space available to store crypt data in */
-	mtr_t * mtr);  /*!< in: mini-transaction */
+fil_space_destroy_crypt_data(
+	fil_space_crypt_t **crypt_data);
 
-/*********************************************************************
-Clear crypt data from page 0 (used for import tablespace) */
+/******************************************************************
+Parse a MLOG_FILE_WRITE_CRYPT_DATA log entry
+@param[in]	ptr		Log entry start
+@param[in]	end_ptr		Log entry end
+@param[in]	block		buffer block
+@return position on log buffer */
 UNIV_INTERN
-void
-fil_space_clear_crypt_data(
-/*=======================*/
-	byte* page,    /*!< in: buffer page */
-	ulint offset); /*!< in: offset where crypt data is stored */
+const byte*
+fil_parse_write_crypt_data(
+	const byte*		ptr,
+	const byte*		end_ptr,
+	const buf_block_t*	block)
+	MY_ATTRIBUTE((warn_unused_result));
 
-/*********************************************************************
-Parse crypt data log record */
+/******************************************************************
+Encrypt a buffer
+@param[in,out]		crypt_data	Crypt data
+@param[in]		space		space_id
+@param[in]		offset		Page offset
+@param[in]		lsn		Log sequence number
+@param[in]		src_frame	Page to encrypt
+@param[in]		zip_size	Compressed size or 0
+@param[in,out]		dst_frame	Output buffer
+@return encrypted buffer or NULL */
 UNIV_INTERN
 byte*
-fil_parse_write_crypt_data(
-/*=======================*/
-	byte* ptr,     /*!< in: start of log record */
-	byte* end_ptr, /*!< in: end of log record */
-	buf_block_t*); /*!< in: buffer page to apply record to */
+fil_encrypt_buf(
+	fil_space_crypt_t* crypt_data,
+	ulint		space,
+	ulint		offset,
+	lsn_t		lsn,
+	const byte*	src_frame,
+	ulint		zip_size,
+	byte*		dst_frame)
+	MY_ATTRIBUTE((warn_unused_result));
 
-/*********************************************************************
-Check if extra buffer shall be allocated for decrypting after read */
+/******************************************************************
+Encrypt a page
+
+@param[in]		space		Tablespace
+@param[in]		offset		Page offset
+@param[in]		lsn		Log sequence number
+@param[in]		src_frame	Page to encrypt
+@param[in,out]		dst_frame	Output buffer
+@return encrypted buffer or NULL */
 UNIV_INTERN
-bool
-fil_space_check_encryption_read(
-/*============================*/
-	ulint space);          /*!< in: tablespace id */
+byte*
+fil_space_encrypt(
+	const fil_space_t* space,
+	ulint		offset,
+	lsn_t		lsn,
+	byte*		src_frame,
+	byte*		dst_frame)
+	MY_ATTRIBUTE((warn_unused_result));
 
 /******************************************************************
 Decrypt a page
+@param[in,out]	crypt_data		crypt_data
+@param[in]	tmp_frame		Temporary buffer
+@param[in]	page_size		Page size
+@param[in,out]	src_frame		Page to decrypt
+@param[out]	err			DB_SUCCESS or error
 @return true if page decrypted, false if not.*/
 UNIV_INTERN
 bool
 fil_space_decrypt(
-/*==============*/
-	fil_space_crypt_t*	crypt_data,	/*!< in: crypt data */
-	byte*			tmp_frame,	/*!< in: temporary buffer */
-	ulint			page_size,	/*!< in: page size */
-	byte*			src_frame,	/*!< in: out: page buffer */
-	dberr_t*		err)		/*!< in: out: DB_SUCCESS or
-						error code */
-	MY_ATTRIBUTE((warn_unused_result));
-
-/*********************************************************************
-Encrypt buffer page
-@return encrypted page, or original not encrypted page if encrypt
-is not needed. */
-UNIV_INTERN
-byte*
-fil_space_encrypt(
-/*==============*/
-	ulint	space,		/*!< in: tablespace id */
-	ulint	offset,		/*!< in: page no */
-	lsn_t	lsn,		/*!< in: page lsn */
-	byte*	src_frame,	/*!< in: page frame */
-	ulint	size,		/*!< in: size of data to encrypt */
-	byte*	dst_frame);	/*!< in: where to encrypt to */
+	fil_space_crypt_t*	crypt_data,
+	byte*			tmp_frame,
+	ulint			page_size,
+	byte*			src_frame,
+	dberr_t*		err);
 
 /******************************************************************
 Decrypt a page
-@param[in]	space			Tablespace id
+@param[in]	space			Tablespace
 @param[in]	tmp_frame		Temporary buffer used for decrypting
 @param[in]	page_size		Page size
 @param[in,out]	src_frame		Page to decrypt
@@ -368,13 +377,24 @@ not needed.*/
 UNIV_INTERN
 byte*
 fil_space_decrypt(
-/*==============*/
-	ulint	space,
-	byte*	src_frame,
-	ulint	page_size,
-	byte*	dst_frame,
-	bool*	decrypted)
-	__attribute__((warn_unused_result));
+	const fil_space_t* space,
+	byte*		tmp_frame,
+	byte*		src_frame,
+	bool*		decrypted)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/******************************************************************
+Calculate post encryption checksum
+@param[in]	zip_size	zip_size or 0
+@param[in]	dst_frame	Block where checksum is calculated
+@return page checksum or BUF_NO_CHECKSUM_MAGIC
+not needed. */
+UNIV_INTERN
+ulint
+fil_crypt_calculate_checksum(
+	ulint		zip_size,
+	const byte*	dst_frame)
+	MY_ATTRIBUTE((warn_unused_result));
 
 /*********************************************************************
 Verify that post encryption checksum match calculated checksum.
@@ -383,6 +403,7 @@ metadata (this is strong indication that tablespace is encrypted).
 Function also verifies that traditional checksum does not match
 calculated checksum as if it does page could be valid unencrypted,
 encrypted, or corrupted.
+
 @param[in]	page		Page to verify
 @param[in]	zip_size	zip size
 @param[in]	space		Tablespace
@@ -391,161 +412,94 @@ encrypted, or corrupted.
 UNIV_INTERN
 bool
 fil_space_verify_crypt_checksum(
-	byte*			page,
+	byte* 			page,
 	ulint			zip_size,
 	const fil_space_t*	space,
 	ulint			pageno)
-	__attribute__((warn_unused_result));
+	MY_ATTRIBUTE((warn_unused_result));
 
 /*********************************************************************
-Init threads for key rotation */
+Adjust thread count for key rotation
+@param[in]	enw_cnt		Number of threads to be used */
 UNIV_INTERN
 void
-fil_crypt_threads_init();
+fil_crypt_set_thread_cnt(
+	uint	new_cnt);
 
 /*********************************************************************
-Set thread count (e.g start or stops threads) used for key rotation */
+Adjust max key age
+@param[in]	val		New max key age */
 UNIV_INTERN
 void
-fil_crypt_set_thread_cnt(
-/*=====================*/
-	uint new_cnt); /*!< in: requested #threads */
+fil_crypt_set_rotate_key_age(
+	uint	val);
 
 /*********************************************************************
-Cleanup resources for threads for key rotation */
+Adjust rotation iops
+@param[in]	val		New max roation iops */
 UNIV_INTERN
 void
-fil_crypt_threads_cleanup();
+fil_crypt_set_rotation_iops(
+	uint val);
 
 /*********************************************************************
-Set rotate key age */
+Adjust encrypt tables
+@param[in]	val		New setting for innodb-encrypt-tables */
 UNIV_INTERN
 void
-fil_crypt_set_rotate_key_age(
-/*=========================*/
-	uint rotate_age); /*!< in: requested rotate age */
+fil_crypt_set_encrypt_tables(
+	uint val);
 
 /*********************************************************************
-Set rotation threads iops */
+Init threads for key rotation */
 UNIV_INTERN
 void
-fil_crypt_set_rotation_iops(
-/*========================*/
-	uint iops); /*!< in: requested iops */
+fil_crypt_threads_init();
 
 /*********************************************************************
-Mark a space as closing */
+Clean up key rotation threads resources */
 UNIV_INTERN
 void
-fil_space_crypt_mark_space_closing(
-/*===============================*/
-	ulint			space,		/*!< in: tablespace id */
-	fil_space_crypt_t*	crypt_data);	/*!< in: crypt_data or NULL */
+fil_crypt_threads_cleanup();
 
 /*********************************************************************
-Wait for crypt threads to stop accessing space */
+Wait for crypt threads to stop accessing space
+@param[in]	space		Tablespace */
 UNIV_INTERN
 void
 fil_space_crypt_close_tablespace(
-/*=============================*/
-	ulint space);          /*!< in: tablespace id */
-
-/** Struct for retreiving info about encryption */
-struct fil_space_crypt_status_t {
-	ulint space;             /*!< tablespace id */
-	ulint scheme;            /*!< encryption scheme */
-	uint  min_key_version;   /*!< min key version */
-	uint  current_key_version;/*!< current key version */
-	uint  keyserver_requests;/*!< no of key requests to key server */
-	ulint key_id;            /*!< current key_id */
-	bool rotating;           /*!< is key rotation ongoing */
-	bool flushing;           /*!< is flush at end of rotation ongoing */
-	ulint rotate_next_page_number; /*!< next page if key rotating */
-	ulint rotate_max_page_number;  /*!< max page if key rotating */
-};
+	const fil_space_t*	space);
 
 /*********************************************************************
-Get crypt status for a space
-@return 0 if crypt data found */
+Get crypt status for a space (used by information_schema)
+@param[in]	space		Tablespace
+@param[out]	status		Crypt status
+return 0 if crypt data present */
 UNIV_INTERN
-int
+void
 fil_space_crypt_get_status(
-/*=======================*/
-	ulint id,	                           /*!< in: space id */
-	struct fil_space_crypt_status_t * status); /*!< out: status  */
-
-/** Struct for retreiving statistics about encryption key rotation */
-struct fil_crypt_stat_t {
-	ulint pages_read_from_cache;
-	ulint pages_read_from_disk;
-	ulint pages_modified;
-	ulint pages_flushed;
-	ulint estimated_iops;
-};
+	const fil_space_t*			space,
+	struct fil_space_crypt_status_t*	status);
 
 /*********************************************************************
-Get crypt rotation statistics */
+Return crypt statistics
+@param[out]	stat		Crypt statistics */
 UNIV_INTERN
 void
 fil_crypt_total_stat(
-/*==================*/
-	fil_crypt_stat_t* stat); /*!< out: crypt stat */
-
-/** Struct for retreiving info about scrubbing */
-struct fil_space_scrub_status_t {
-	ulint space;             /*!< tablespace id */
-	bool compressed;        /*!< is space compressed  */
-	time_t last_scrub_completed;  /*!< when was last scrub completed */
-	bool scrubbing;               /*!< is scrubbing ongoing */
-	time_t current_scrub_started; /*!< when started current scrubbing */
-	ulint current_scrub_active_threads; /*!< current scrub active threads */
-	ulint current_scrub_page_number; /*!< current scrub page no */
-	ulint current_scrub_max_page_number; /*!< current scrub max page no */
-};
+	fil_crypt_stat_t *stat);
 
 /*********************************************************************
-Get scrub status for a space
-@return 0 if no scrub info found */
-UNIV_INTERN
-int
-fil_space_get_scrub_status(
-/*=======================*/
-	ulint id,	                           /*!< in: space id */
-	struct fil_space_scrub_status_t * status); /*!< out: status  */
+Get scrub status for a space (used by information_schema)
 
-/*********************************************************************
-Adjust encrypt tables */
+@param[in]	space		Tablespace
+@param[out]	status		Scrub status
+return 0 if data found */
 UNIV_INTERN
 void
-fil_crypt_set_encrypt_tables(
-/*=========================*/
-	uint val);      /*!< in: New srv_encrypt_tables setting */
-
-/******************************************************************
-Encrypt a buffer */
-UNIV_INTERN
-byte*
-fil_encrypt_buf(
-/*============*/
-	fil_space_crypt_t* crypt_data,	/*!< in: crypt data */
-	ulint		space,		/*!< in: Space id */
-	ulint		offset,		/*!< in: Page offset */
-	lsn_t		lsn,		/*!< in: lsn */
-	byte*		src_frame,	/*!< in: Source page to be encrypted */
-	ulint		zip_size,	/*!< in: compressed size if
-					row_format compressed */
-	byte*		dst_frame);	/*!< in: outbut buffer */
-
-/******************************************************************
-Calculate post encryption checksum
-@return page checksum or BUF_NO_CHECKSUM_MAGIC
-not needed. */
-UNIV_INTERN
-ulint
-fil_crypt_calculate_checksum(
-/*=========================*/
-	ulint	zip_size,	/*!< in: zip_size or 0 */
-	byte*	dst_frame);	/*!< in: page where to calculate */
+fil_space_get_scrub_status(
+	const fil_space_t*			space,
+	struct fil_space_scrub_status_t*	status);
 
 #ifndef UNIV_NONINL
 #include "fil0crypt.ic"
diff --git a/storage/innobase/include/fil0crypt.ic b/storage/innobase/include/fil0crypt.ic
index 5fafa6cd3f0..cb9ba083466 100644
--- a/storage/innobase/include/fil0crypt.ic
+++ b/storage/innobase/include/fil0crypt.ic
@@ -34,35 +34,3 @@ fil_page_is_encrypted(
 {
 	return(mach_read_from_4(buf+FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION) != 0);
 }
-
-/*******************************************************************//**
-Find out whether the page can be decrypted.
-The function for decrypting the page should already be executed before this.
-@return	1 if key provider not available or key is not available
-	0 if decryption should be possible
-*/
-UNIV_INLINE
-bool
-fil_page_encryption_status(
-/*===================*/
-	const byte *buf,	/*!< in: page */
-	ulint space_id)		/*!< in: space_id */
-{
-	fil_space_crypt_t *crypt_data = fil_space_get_crypt_data(space_id);
-	ulint page_type = mach_read_from_2(buf+FIL_PAGE_TYPE);
-
-	if (page_type == FIL_PAGE_TYPE_FSP_HDR) {
-		if (crypt_data != NULL) {
-			if (!encryption_key_id_exists(crypt_data->key_id)) {
-				/* accessing table would surely fail, because no key or no key provider available */
-				return 1;
-			}
-		}
-	} else {
-		ulint key = mach_read_from_4(buf + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
-		if (!encryption_key_version_exists(crypt_data->key_id, key)) {
-			return 1;
-		}
-	}
-	return 0;
-}
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index cf9297c93ae..b4ee931fdbe 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -185,8 +185,18 @@ extern fil_addr_t	fil_addr_null;
 
 #ifndef UNIV_INNOCHECKSUM
 
-/* structure containing encryption specification */
-typedef struct fil_space_crypt_struct fil_space_crypt_t;
+/** Structure containing encryption specification */
+struct fil_space_crypt_t;
+
+/** Enum values for encryption table option */
+enum fil_encryption_t {
+	/** Encrypted if innodb_encrypt_tables=ON (srv_encrypt_tables) */
+	FIL_ENCRYPTION_DEFAULT,
+	/** Encrypted */
+	FIL_ENCRYPTION_ON,
+	/** Not encrypted */
+	FIL_ENCRYPTION_OFF
+};
 
 /** The number of fsyncs done to the log */
 extern ulint	fil_n_log_flushes;
@@ -276,7 +286,7 @@ struct fil_space_t {
 				.ibd file of tablespace and want to
 				stop temporarily posting of new i/o
 				requests on the file */
-	ibool		stop_new_ops;
+	bool		stop_new_ops;
 				/*!< we set this TRUE when we start
 				deleting a single-table tablespace.
 				When this is set following new ops
@@ -343,7 +353,22 @@ struct fil_space_t {
 	UT_LIST_NODE_T(fil_space_t) space_list;
 				/*!< list of all spaces */
 
+	/*!< Protected by fil_system */
+	UT_LIST_NODE_T(fil_space_t) rotation_list;
+				/*!< list of spaces needing
+				key rotation */
+
+	bool		is_in_rotation_list;
+				/*!< true if this space is
+				currently in key rotation list */
+
 	ulint		magic_n;/*!< FIL_SPACE_MAGIC_N */
+
+	/** @return whether the tablespace is about to be dropped or truncated */
+	bool is_stopping() const
+	{
+		return stop_new_ops;
+	}
 };
 
 /** Value of fil_space_t::magic_n */
@@ -399,6 +424,11 @@ struct fil_system_t {
 					request */
 	UT_LIST_BASE_NODE_T(fil_space_t) space_list;
 					/*!< list of all file spaces */
+
+	UT_LIST_BASE_NODE_T(fil_space_t) rotation_list;
+					/*!< list of all file spaces needing
+					key rotation.*/
+
 	ibool		space_id_reuse_warned;
 					/* !< TRUE if fil_space_create()
 					has issued a warning about
@@ -470,18 +500,24 @@ fil_space_truncate_start(
 /*******************************************************************//**
 Creates a space memory object and puts it to the 'fil system' hash table.
 If there is an error, prints an error message to the .err log.
+@param[in]	name		Space name
+@param[in]	id		Space id
+@param[in]	flags		Tablespace flags
+@param[in]	purpose		FIL_TABLESPACE or FIL_LOG if log
+@param[in]	crypt_data	Encryption information
+@param[in]	create_table	True if this is create table
+@param[in]	mode		Encryption mode
 @return	TRUE if success */
 UNIV_INTERN
-ibool
+bool
 fil_space_create(
-/*=============*/
-	const char*	name,	/*!< in: space name */
-	ulint		id,	/*!< in: space id */
-	ulint		zip_size,/*!< in: compressed page size, or
-				0 for uncompressed tablespaces */
-	ulint		purpose, /*!< in: FIL_TABLESPACE, or FIL_LOG if log */
-	fil_space_crypt_t* crypt_data, /*!< in: crypt data */
-	bool		create_table); /*!< in: true if create table */
+	const char*		name,
+	ulint			id,
+	ulint			flags,
+	ulint			purpose,
+	fil_space_crypt_t*	crypt_data,
+	bool			create_table,
+	fil_encryption_t	mode = FIL_ENCRYPTION_DEFAULT);
 
 /*******************************************************************//**
 Assigns a new space id for a new single-table tablespace. This works simply by
@@ -604,6 +640,59 @@ fil_write_flushed_lsn_to_data_files(
 /*================================*/
 	lsn_t	lsn,		/*!< in: lsn to write */
 	ulint	arch_log_no);	/*!< in: latest archived log file number */
+
+/** Acquire a tablespace when it could be dropped concurrently.
+Used by background threads that do not necessarily hold proper locks
+for concurrency control.
+@param[in]	id	tablespace ID
+@return the tablespace, or NULL if missing or being deleted */
+fil_space_t*
+fil_space_acquire(
+	ulint	id)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/** Acquire a tablespace that may not exist.
+Used by background threads that do not necessarily hold proper locks
+for concurrency control.
+@param[in]	id	tablespace ID
+@return the tablespace, or NULL if missing or being deleted */
+fil_space_t*
+fil_space_acquire_silent(
+	ulint	id)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/** Release a tablespace acquired with fil_space_acquire().
+@param[in,out]	space	tablespace to release  */
+void
+fil_space_release(
+	fil_space_t*	space);
+
+/** Return the next fil_space_t.
+Once started, the caller must keep calling this until it returns NULL.
+fil_space_acquire() and fil_space_release() are invoked here which
+blocks a concurrent operation from dropping the tablespace.
+@param[in,out]	prev_space	Pointer to the previous fil_space_t.
+If NULL, use the first fil_space_t on fil_system->space_list.
+@return pointer to the next fil_space_t.
+@retval NULL if this was the last  */
+fil_space_t*
+fil_space_next(
+	fil_space_t*	prev_space)
+	MY_ATTRIBUTE((warn_unused_result));
+
+/** Return the next fil_space_t from key rotation list.
+Once started, the caller must keep calling this until it returns NULL.
+fil_space_acquire() and fil_space_release() are invoked here which
+blocks a concurrent operation from dropping the tablespace.
+@param[in,out]	prev_space	Pointer to the previous fil_space_t.
+If NULL, use the first fil_space_t on fil_system->space_list.
+@return pointer to the next fil_space_t.
+@retval NULL if this was the last*/
+fil_space_t*
+fil_space_keyrotate_next(
+	fil_space_t*	prev_space)
+	MY_ATTRIBUTE((warn_unused_result));
+
 /*******************************************************************//**
 Reads the flushed lsn, arch no, and tablespace flag fields from a data
 file at database startup.
diff --git a/storage/innobase/include/fsp0fsp.h b/storage/innobase/include/fsp0fsp.h
index 24fe382b344..2a162100dc1 100644
--- a/storage/innobase/include/fsp0fsp.h
+++ b/storage/innobase/include/fsp0fsp.h
@@ -1038,14 +1038,15 @@ fsp_flags_get_page_size(
 /*====================*/
 	ulint	flags);		/*!< in: tablespace flags */
 
-/*********************************************************************/
-/* @return offset into fsp header where crypt data is stored */
+/*********************************************************************
+Compute offset after xdes where crypt data can be stored
+@param[in]	zip_size	Compressed size or 0
+@return	offset */
 UNIV_INTERN
 ulint
 fsp_header_get_crypt_offset(
-/*========================*/
-	ulint zip_size,		/*!< in: zip_size */
-	ulint* max_size);	/*!< out: free space after offset */
+	const ulint zip_size)
+	MY_ATTRIBUTE((warn_unused_result));
 
 #define fsp_page_is_free(space,page,mtr) \
 	fsp_page_is_free_func(space,page,mtr, __FILE__, __LINE__)
diff --git a/storage/innobase/include/srv0srv.h b/storage/innobase/include/srv0srv.h
index c615c837c04..9753c442727 100644
--- a/storage/innobase/include/srv0srv.h
+++ b/storage/innobase/include/srv0srv.h
@@ -185,6 +185,9 @@ struct srv_stats_t {
 
 	/** Number of encryption_get_latest_key_version calls */
 	ulint_ctr_64_t		n_key_requests;
+
+	/** Number of spaces in keyrotation list */
+	ulint_ctr_64_t		key_rotation_list_length;
 };
 
 extern const char*	srv_main_thread_op_info;
@@ -1032,6 +1035,7 @@ struct export_var_t{
 	ulint innodb_encryption_rotation_pages_flushed;
 	ulint innodb_encryption_rotation_estimated_iops;
 	ib_int64_t innodb_encryption_key_requests;
+	ib_int64_t innodb_key_rotation_list_length;
 
 	ulint innodb_scrub_page_reorganizations;
 	ulint innodb_scrub_page_splits;
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 532ad2a3f76..4812dc9ae91 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2014, 2016, MariaDB Corporation
+Copyright (c) 2014, 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
@@ -6398,12 +6398,13 @@ loop:
 
 	if (lock_get_type_low(lock) == LOCK_REC) {
 		if (load_page_first) {
-			ulint	space	= lock->un_member.rec_lock.space;
-			ulint	zip_size= fil_space_get_zip_size(space);
+			ulint	space_id = lock->un_member.rec_lock.space;
+			/* Check if the space is exists or not. only
+			when the space is valid, try to get the page. */
+			fil_space_t* space = fil_space_acquire(space_id);
 			ulint	page_no = lock->un_member.rec_lock.page_no;
-			ibool	tablespace_being_deleted = FALSE;
 
-			if (UNIV_UNLIKELY(zip_size == ULINT_UNDEFINED)) {
+			if (!space) {
 
 				/* It is a single table tablespace and
 				the .ibd file is missing (TRUNCATE
@@ -6412,37 +6413,33 @@ loop:
 				load the page in the buffer pool. */
 
 				fprintf(file, "RECORD LOCKS on"
-					" non-existing space %lu\n",
-					(ulong) space);
+					" non-existing space: " ULINTPF "\n",
+					space_id);
 				goto print_rec;
 			}
 
+			const ulint zip_size = fsp_flags_get_zip_size(space->flags);
+
 			lock_mutex_exit();
 			mutex_exit(&trx_sys->mutex);
 
 			DEBUG_SYNC_C("innodb_monitor_before_lock_page_read");
 
-			/* Check if the space is exists or not. only when the space
-			is valid, try to get the page. */
-			tablespace_being_deleted = fil_inc_pending_ops(space, false);
-
-			if (!tablespace_being_deleted) {
+			if (space) {
 				mtr_start(&mtr);
 
-				buf_page_get_gen(space, zip_size, page_no,
+				buf_page_get_gen(space_id, zip_size, page_no,
 						 RW_NO_LATCH, NULL,
 						 BUF_GET_POSSIBLY_FREED,
 						 __FILE__, __LINE__, &mtr);
 
 				mtr_commit(&mtr);
 
-				fil_decr_pending_ops(space);
-			} else {
-				fprintf(file, "RECORD LOCKS on"
-					" non-existing space %lu\n",
-					(ulong) space);
 			}
 
+
+			fil_space_release(space);
+
 			load_page_first = FALSE;
 
 			lock_mutex_enter();
@@ -6868,7 +6865,7 @@ static
 void
 lock_rec_block_validate(
 /*====================*/
-	ulint		space,
+	ulint		space_id,
 	ulint		page_no)
 {
 	/* The lock and the block that it is referring to may be freed at
@@ -6881,10 +6878,11 @@ lock_rec_block_validate(
 
 	/* Make sure that the tablespace is not deleted while we are
 	trying to access the page. */
-	if (!fil_inc_pending_ops(space, true)) {
+	if (fil_space_t* space = fil_space_acquire(space_id)) {
+
 		mtr_start(&mtr);
 		block = buf_page_get_gen(
-			space, fil_space_get_zip_size(space),
+			space_id, fsp_flags_get_zip_size(space->flags),
 			page_no, RW_X_LATCH, NULL,
 			BUF_GET_POSSIBLY_FREED,
 			__FILE__, __LINE__, &mtr);
@@ -6894,7 +6892,7 @@ lock_rec_block_validate(
 		ut_ad(lock_rec_validate_page(block));
 		mtr_commit(&mtr);
 
-		fil_decr_pending_ops(space);
+		fil_space_release(space);
 	}
 }
 
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index d088ae9f3df..45fd42848e8 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -1297,7 +1297,7 @@ recv_parse_or_apply_log_rec_body(
 		}
 		break;
 	case MLOG_FILE_WRITE_CRYPT_DATA:
-		ptr = fil_parse_write_crypt_data(ptr, end_ptr, block);
+		ptr = const_cast<byte*>(fil_parse_write_crypt_data(ptr, end_ptr, block));
 		break;
 	default:
 		ptr = NULL;
diff --git a/storage/innobase/pars/pars0pars.cc b/storage/innobase/pars/pars0pars.cc
index e6af3d25e86..ce61d6e1e3b 100644
--- a/storage/innobase/pars/pars0pars.cc
+++ b/storage/innobase/pars/pars0pars.cc
@@ -2020,7 +2020,7 @@ pars_create_table(
 	}
 
 	node = tab_create_graph_create(table, pars_sym_tab_global->heap, true,
-		FIL_SPACE_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+		FIL_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
 
 	table_sym->resolved = TRUE;
 	table_sym->token_type = SYM_TABLE;
diff --git a/storage/innobase/row/row0ftsort.cc b/storage/innobase/row/row0ftsort.cc
index e8e02cdcc86..0c1ff402f8f 100644
--- a/storage/innobase/row/row0ftsort.cc
+++ b/storage/innobase/row/row0ftsort.cc
@@ -221,7 +221,14 @@ row_fts_psort_info_init(
 	common_info->sort_event = os_event_create();
 	common_info->merge_event = os_event_create();
 	common_info->opt_doc_id_size = opt_doc_id_size;
-	crypt_data = fil_space_get_crypt_data(new_table->space);
+
+	/* Theoretically the tablespace can be dropped straight away.
+	In practice, the DDL completion will wait for this thread to
+	finish. */
+	if (fil_space_t* space = fil_space_acquire(new_table->space)) {
+		crypt_data = space->crypt_data;
+		fil_space_release(space);
+	}
 
 	if (crypt_data && crypt_data->should_encrypt()) {
 		common_info->crypt_data = crypt_data;
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 15d2ade303d..5531263b248 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -3948,7 +3948,7 @@ row_merge_build_indexes(
 {
 	merge_file_t*		merge_files;
 	row_merge_block_t*	block;
-	row_merge_block_t*	crypt_block;
+	row_merge_block_t*	crypt_block = NULL;
 	ulint			block_size;
 	ulint			i;
 	ulint			j;
@@ -3984,9 +3984,15 @@ row_merge_build_indexes(
 		DBUG_RETURN(DB_OUT_OF_MEMORY);
 	}
 
-	/* Get crypt data from tablespace if present. */
-	crypt_data = fil_space_get_crypt_data(new_table->space);
-	crypt_block = NULL;
+	/* Get crypt data from tablespace if present. We should be protected
+	from concurrent DDL (e.g. drop table) by MDL-locks. */
+	fil_space_t* space = fil_space_acquire(new_table->space);
+
+	if (space) {
+		crypt_data = space->crypt_data;
+	} else {
+		DBUG_RETURN(DB_TABLESPACE_NOT_FOUND);
+	}
 
 	/* If tablespace is encrypted, allocate additional buffer for
 	encryption/decryption. */
@@ -4350,5 +4356,9 @@ func_exit:
 		}
 	}
 
+	if (space) {
+		fil_space_release(space);
+	}
+
 	DBUG_RETURN(error);
 }
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index d0088802adf..b479697d644 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -3289,21 +3289,17 @@ void
 fil_wait_crypt_bg_threads(
 	dict_table_t* table)
 {
-	uint start = time(0);
-	uint last = start;
-
-	if (table->space != 0) {
-		fil_space_crypt_mark_space_closing(table->space, table->crypt_data);
-	}
+	time_t start = time(0);
+	time_t last = start;
 
 	while (table->n_ref_count > 0) {
 		dict_mutex_exit_for_mysql();
 		os_thread_sleep(20000);
 		dict_mutex_enter_for_mysql();
-		uint now = time(0);
+		time_t now = time(0);
 		if (now >= last + 30) {
 			fprintf(stderr,
-				"WARNING: waited %u seconds "
+				"WARNING: waited %ld seconds "
 				"for ref-count on table: %s space: %u\n",
 				now - start, table->name, table->space);
 			last = now;
@@ -3311,7 +3307,7 @@ fil_wait_crypt_bg_threads(
 
 		if (now >= start + 300) {
 			fprintf(stderr,
-				"WARNING: after %u seconds, gave up waiting "
+				"WARNING: after %ld seconds, gave up waiting "
 				"for ref-count on table: %s space: %u\n",
 				now - start, table->name, table->space);
 			break;
@@ -3507,35 +3503,40 @@ row_truncate_table_for_mysql(
 
 	if (table->space && !DICT_TF2_FLAG_IS_SET(table, DICT_TF2_TEMPORARY)) {
 		/* Discard and create the single-table tablespace. */
-		fil_space_crypt_t* crypt_data;
-		ulint	space	= table->space;
-		ulint	flags	= fil_space_get_flags(space);
+		ulint	space_id = table->space;
+		ulint	flags	= ULINT_UNDEFINED;
 		ulint	key_id  = FIL_DEFAULT_ENCRYPTION_KEY;
-		fil_encryption_t mode = FIL_SPACE_ENCRYPTION_DEFAULT;
+		fil_encryption_t mode = FIL_ENCRYPTION_DEFAULT;
 
 		dict_get_and_save_data_dir_path(table, true);
-		crypt_data = fil_space_get_crypt_data(space);
 
-		if (crypt_data) {
-			key_id = crypt_data->key_id;
-			mode = crypt_data->encryption;
+		if (fil_space_t* space = fil_space_acquire(space_id)) {
+			fil_space_crypt_t* crypt_data = space->crypt_data;
+
+			if (crypt_data) {
+				key_id = crypt_data->key_id;
+				mode = crypt_data->encryption;
+			}
+
+			flags = space->flags;
+			fil_space_release(space);
 		}
 
 		if (flags != ULINT_UNDEFINED
-		    && fil_discard_tablespace(space) == DB_SUCCESS) {
+		    && fil_discard_tablespace(space_id) == DB_SUCCESS) {
 
 			dict_index_t*	index;
 
-			dict_hdr_get_new_id(NULL, NULL, &space);
+			dict_hdr_get_new_id(NULL, NULL, &space_id);
 
 			/* Lock all index trees for this table. We must
 			do so after dict_hdr_get_new_id() to preserve
 			the latch order */
 			dict_table_x_lock_indexes(table);
 
-			if (space == ULINT_UNDEFINED
+			if (space_id == ULINT_UNDEFINED
 			    || fil_create_new_single_table_tablespace(
-				    space, table->name,
+				    space_id, table->name,
 				    table->data_dir_path,
 				    flags, table->flags2,
 				    FIL_IBD_FILE_INITIAL_SIZE,
@@ -3553,21 +3554,21 @@ row_truncate_table_for_mysql(
 				goto funct_exit;
 			}
 
-			recreate_space = space;
+			recreate_space = space_id;
 
 			/* Replace the space_id in the data dictionary cache.
 			The persisent data dictionary (SYS_TABLES.SPACE
 			and SYS_INDEXES.SPACE) are updated later in this
 			function. */
-			table->space = space;
+			table->space = space_id;
 			index = dict_table_get_first_index(table);
 			do {
-				index->space = space;
+				index->space = space_id;
 				index = dict_table_get_next_index(index);
 			} while (index);
 
 			mtr_start_trx(&mtr, trx);
-			fsp_header_init(space,
+			fsp_header_init(space_id,
 					FIL_IBD_FILE_INITIAL_SIZE, &mtr);
 			mtr_commit(&mtr);
 		}
@@ -4235,7 +4236,13 @@ row_drop_table_for_mysql(
 	/* If table has not yet have crypt_data, try to read it to
 	make freeing the table easier. */
 	if (!table->crypt_data) {
-		table->crypt_data = fil_space_get_crypt_data(table->space);
+
+		if (fil_space_t* space = fil_space_acquire_silent(table->space)) {
+			/* We use crypt data in dict_table_t in ha_innodb.cc
+			to push warnings to user thread. */
+			table->crypt_data = space->crypt_data;
+			fil_space_release(space);
+		}
 	}
 
 	/* We use the private SQL parser of Innobase to generate the
diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
index a02b522b116..8f8e4e24721 100644
--- a/storage/innobase/srv/srv0srv.cc
+++ b/storage/innobase/srv/srv0srv.cc
@@ -1689,6 +1689,8 @@ srv_export_innodb_status(void)
 		crypt_stat.estimated_iops;
 	export_vars.innodb_encryption_key_requests =
 		srv_stats.n_key_requests;
+	export_vars.innodb_key_rotation_list_length =
+		srv_stats.key_rotation_list_length;
 
 	export_vars.innodb_scrub_page_reorganizations =
 		scrub_stat.page_reorganizations;
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 6fe2c2a0f2a..048fdb8cafd 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -665,6 +665,7 @@ create_log_files(
 		FIL_LOG,
 		NULL /* no encryption yet */,
 		true /* this is create */);
+
 	ut_a(fil_validate());
 
 	logfile0 = fil_node_create(
@@ -1153,13 +1154,14 @@ check_first_page:
 
 		if (i == 0) {
 			if (!crypt_data) {
-				crypt_data = fil_space_create_crypt_data(FIL_SPACE_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+				crypt_data = fil_space_create_crypt_data(FIL_ENCRYPTION_DEFAULT,
+					FIL_DEFAULT_ENCRYPTION_KEY);
 			}
 
 			flags = FSP_FLAGS_PAGE_SSIZE();
 
 			fil_space_create(name, 0, flags, FIL_TABLESPACE,
-					crypt_data, (*create_new_db) == true);
+				crypt_data, (*create_new_db) == true);
 		}
 
 		ut_a(fil_validate());