MDEV-14398 When innodb_encryption_rotate_key_age=0 is set,bb-10.1-mdev-14398

			server won't encrypt tablespaces.

The following commit makes the dependency between
innodb-encryption-rotate-key-age and encryption/decryption.

commit 50eb40a2a8aa3af6cc271f6028f4d6d74301d030
Author: Jan Lindström <jan.lindstrom@mariadb.com>
Date: Tue Mar 14 12:56:01 2017 +0200

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.

The above commit basically make use of innodb-encryption-rotate-key-age=0
to disable key rotation and background data scrubbing.
It only encrypts/decrypts for the newly created table.

One more issue is with above commit is that, even though the
innodb_encrypt_table value is changed at run time. It doesn't affect
the state of the old tablespace.

Different Solution:
===================

(1) Key rotation thread can iterate through tablespace list and check whether
it matches with innodb_encrypt_tables parameter. If it doesn't match then do
the operation based on the innodb_encrypt_tables parameter. InnoDB shouldn't
iterate the fil_system->space_list again and again. It increases the CPU usage.

(2) Key rotation thread should store lastly used tablespace id. If the value of
innodb-encryption-rotate-key-age is set to 0 then InnoDB should
start the traversion from lastly used space id. So that key rotation thread
only does the encryption/decryption for newly added tablespace.

    (i) If lastly used tablespace id is not found or dropped then InnoDB
	should traverse it from system tablespace again.

(3) If innodb_encrypt_tables value changed at runtime then InnoDB should
traverse the tablespace from system tablespace again.

8 files changed, 54 insertions, 257 deletions

diff --git a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
index 6c09e015a1e..1a16d966ea4 100644
--- a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
+++ b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
@@ -1,7 +1,5 @@
 SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
 NAME
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
-NAME
 create database enctests;
 use enctests;
 create table t1(a int not null primary key, b char(200)) engine=innodb;
diff --git a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
index 574e0c3becc..8a22043f660 100644
--- a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
+++ b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
@@ -4,7 +4,6 @@
 -- source include/not_embedded.inc
 
 SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
 
 --disable_query_log
 --disable_warnings
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 2d149f07433..c5df83f0e09 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -1175,6 +1175,9 @@ struct rotate_thread_t {
 		thread_no = no;
 		first = true;
 		estimated_max_iops = 20;
+		last_space_id = 0;
+		old_encrypt_tables = srv_encrypt_tables;
+		space = NULL;
 	}
 
 	uint thread_no;
@@ -1194,7 +1197,9 @@ struct rotate_thread_t {
 
 	btr_scrub_t scrub_data;      /* thread local data used by btr_scrub-functions
 				     * when iterating pages of tablespace */
+	ulint last_space_id;	/*!< lastly processed space id */
 
+	ulong old_encrypt_tables; /*!< value of innodb_encrypt_tables variable */
 	/** @return whether this thread should terminate */
 	bool should_shutdown() const {
 		switch (srv_shutdown_state) {
@@ -1542,13 +1547,27 @@ fil_crypt_find_space_to_rotate(
 		state->space = NULL;
 	}
 
-	/* 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) {
+	/* If innodb_encrypt_tables value changed at runtime then
+	start the processing from system tablespace again. */
+	if (state->old_encrypt_tables != srv_encrypt_tables) {
+		state->last_space_id = 0;
+		state->old_encrypt_tables = srv_encrypt_tables;
+		state->space = NULL;
+	}
+
+	/* If innodb-encryption-rotate-key-age value is zero then traverse
+	from last used space id in the list. If the lastly used space is
+	dropped or not found then traverse from system tablespace. */
+	if (srv_fil_crypt_rotate_key_age == 0 && state->last_space_id != 0
+	    && (state->space == NULL || state->space->id != state->last_space_id)) {
+
+		state->space = fil_space_acquire_silent(state->last_space_id);
 		state->space = fil_space_next(state->space);
+		if (state->space) {
+			state->last_space_id = state->space->id;
+		}
 	} else {
-		state->space = fil_space_keyrotate_next(state->space);
+		state->space = fil_space_next(state->space);
 	}
 
 	while (!state->should_shutdown() && state->space) {
@@ -1562,11 +1581,8 @@ fil_crypt_find_space_to_rotate(
 			return true;
 		}
 
-		if (srv_fil_crypt_rotate_key_age) {
-			state->space = fil_space_next(state->space);
-		} else {
-			state->space = fil_space_keyrotate_next(state->space);
-		}
+		state->last_space_id = state->space->id;
+		state->space = fil_space_next(state->space);
 	}
 
 	/* if we didn't find any space return iops */
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 112320bd3f5..097f250afe4 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -1390,22 +1390,7 @@ fil_space_create(
 	UT_LIST_ADD_LAST(space_list, fil_system->space_list, space);
 
 
-	/* 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);
-		space->is_in_rotation_list = true;
-		mutex_exit(&fil_system->mutex);
-		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);
-	}
+	mutex_exit(&fil_system->mutex);
 
 	return(true);
 }
@@ -1514,12 +1499,6 @@ fil_space_free_and_mutex_exit(ulint id, bool x_latched)
 			       space);
 	}
 
-	if (space->is_in_rotation_list) {
-		space->is_in_rotation_list = false;
-		ut_a(UT_LIST_GET_LEN(fil_system->rotation_list) > 0);
-		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);
@@ -6526,89 +6505,3 @@ fil_space_next(fil_space_t* prev_space)
 
 	return(space);
 }
-
-/**
-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)
-{
-	ut_ad(mutex_own(&fil_system->mutex));
-	ut_ad(space);
-
-	if (space->n_pending_ops == 0 && space->is_in_rotation_list) {
-		space->is_in_rotation_list = false;
-		ut_a(UT_LIST_GET_LEN(fil_system->rotation_list) > 0);
-		UT_LIST_REMOVE(rotation_list, fil_system->rotation_list, space);
-	}
-}
-
-
-/** 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*/
-UNIV_INTERN
-fil_space_t*
-fil_space_keyrotate_next(
-	fil_space_t*	prev_space)
-{
-	fil_space_t* space = prev_space;
-	fil_space_t* old   = NULL;
-
-	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);
-		}
-		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->is_stopping())) {
-
-		old = space;
-		space = UT_LIST_GET_NEXT(rotation_list, space);
-		fil_space_remove_from_keyrotation(old);
-	}
-
-	if (space != NULL) {
-		space->n_pending_ops++;
-	}
-
-	mutex_exit(&fil_system->mutex);
-
-	return(space);
-}
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 9b6178bb8f1..3c63fced644 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -353,15 +353,6 @@ 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 */
diff --git a/storage/xtradb/fil/fil0crypt.cc b/storage/xtradb/fil/fil0crypt.cc
index 2d149f07433..3ea5ade33d8 100644
--- a/storage/xtradb/fil/fil0crypt.cc
+++ b/storage/xtradb/fil/fil0crypt.cc
@@ -1175,6 +1175,8 @@ struct rotate_thread_t {
 		thread_no = no;
 		first = true;
 		estimated_max_iops = 20;
+		last_space_id = 0;
+		old_encrypt_tables = srv_encrypt_tables;
 	}
 
 	uint thread_no;
@@ -1194,6 +1196,9 @@ struct rotate_thread_t {
 
 	btr_scrub_t scrub_data;      /* thread local data used by btr_scrub-functions
 				     * when iterating pages of tablespace */
+	ulint last_space_id;	/*!< lastly used space id. */
+
+	ulong old_encrypt_tables; /*!< value of innodb_encrypt_tables variable. */
 
 	/** @return whether this thread should terminate */
 	bool should_shutdown() const {
@@ -1542,13 +1547,27 @@ fil_crypt_find_space_to_rotate(
 		state->space = NULL;
 	}
 
-	/* 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) {
+	/* If innodb_encrypt_tables value changed at runtime then
+	start the processing from system tablespace again. */
+	if (state->old_encrypt_tables != srv_encrypt_tables) {
+		state->last_space_id = 0;
+		state->old_encrypt_tables = srv_encrypt_tables;
+		state->space = NULL;
+	}
+
+	/* If innodb-encryption-rotate-key-age value is zero then traverse
+	from last used space id in the list. If the lastly used space is
+	dropped or not found then traverse from system tablespace. */
+	if (srv_fil_crypt_rotate_key_age == 0 && state->last_space_id != 0
+	    && (state->space == NULL || state->space->id != state->last_space_id)) {
+		state->space = fil_space_acquire_silent(state->last_space_id);
 		state->space = fil_space_next(state->space);
+
+		if (state->space) {
+			state->last_space_id = state->space->id;
+		}
 	} else {
-		state->space = fil_space_keyrotate_next(state->space);
+		state->space = fil_space_next(state->space);
 	}
 
 	while (!state->should_shutdown() && state->space) {
@@ -1562,11 +1581,8 @@ fil_crypt_find_space_to_rotate(
 			return true;
 		}
 
-		if (srv_fil_crypt_rotate_key_age) {
-			state->space = fil_space_next(state->space);
-		} else {
-			state->space = fil_space_keyrotate_next(state->space);
-		}
+		state->last_space_id = state->space->id;
+		state->space = fil_space_next(state->space);
 	}
 
 	/* if we didn't find any space return iops */
diff --git a/storage/xtradb/fil/fil0fil.cc b/storage/xtradb/fil/fil0fil.cc
index bd7dfdf5ac8..00c729cdc23 100644
--- a/storage/xtradb/fil/fil0fil.cc
+++ b/storage/xtradb/fil/fil0fil.cc
@@ -1429,22 +1429,7 @@ fil_space_create(
 
 	UT_LIST_ADD_LAST(space_list, fil_system->space_list, space);
 
-	/* 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);
-		space->is_in_rotation_list = true;
-		mutex_exit(&fil_system->mutex);
-		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);
-	}
+	mutex_exit(&fil_system->mutex);
 
 	return(true);
 }
@@ -1553,12 +1538,6 @@ fil_space_free_and_mutex_exit(ulint id, bool x_latched)
 			       space);
 	}
 
-	if (space->is_in_rotation_list) {
-		space->is_in_rotation_list = false;
-		ut_a(UT_LIST_GET_LEN(fil_system->rotation_list) > 0);
-		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);
@@ -6919,89 +6898,3 @@ fil_space_next(fil_space_t* prev_space)
 
 	return(space);
 }
-
-/**
-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)
-{
-	ut_ad(mutex_own(&fil_system->mutex));
-	ut_ad(space);
-
-	if (space->n_pending_ops == 0 && space->is_in_rotation_list) {
-		space->is_in_rotation_list = false;
-		ut_a(UT_LIST_GET_LEN(fil_system->rotation_list) > 0);
-		UT_LIST_REMOVE(rotation_list, fil_system->rotation_list, space);
-	}
-}
-
-
-/** 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*/
-UNIV_INTERN
-fil_space_t*
-fil_space_keyrotate_next(
-	fil_space_t*	prev_space)
-{
-	fil_space_t* space = prev_space;
-	fil_space_t* old   = NULL;
-
-	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);
-		}
-		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->is_stopping())) {
-
-		old = space;
-		space = UT_LIST_GET_NEXT(rotation_list, space);
-		fil_space_remove_from_keyrotation(old);
-	}
-
-	if (space != NULL) {
-		space->n_pending_ops++;
-	}
-
-	mutex_exit(&fil_system->mutex);
-
-	return(space);
-}
diff --git a/storage/xtradb/include/fil0fil.h b/storage/xtradb/include/fil0fil.h
index a0db48f7037..15a0eb51a23 100644
--- a/storage/xtradb/include/fil0fil.h
+++ b/storage/xtradb/include/fil0fil.h
@@ -352,15 +352,6 @@ 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 */