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.

1 files changed, 49 insertions, 29 deletions

diff --git a/storage/xtradb/buf/buf0buf.cc b/storage/xtradb/buf/buf0buf.cc
index 84e3412387d..5016445b8e7 100644
--- a/storage/xtradb/buf/buf0buf.cc
+++ b/storage/xtradb/buf/buf0buf.cc
@@ -692,7 +692,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
@@ -4571,11 +4571,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
@@ -4605,7 +4608,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);
@@ -4615,7 +4618,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.",
@@ -4627,6 +4630,10 @@ buf_page_check_corrupt(
 		}
 	}
 
+	if (space) {
+		fil_space_release(space);
+	}
+
 	return corrupted;
 }
 
@@ -6248,13 +6255,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);
@@ -6271,7 +6271,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;
 	}
@@ -6288,11 +6301,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;
@@ -6302,11 +6321,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);
@@ -6344,11 +6362,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);
 		}
 
@@ -6359,6 +6376,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;
 }
@@ -6384,16 +6402,6 @@ 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);
-
-	/* Page is encrypted if encryption information is found from
-	tablespace and page contains used key_version. This is true
-	also for pages first compressed and then encrypted. */
-	if (!crypt_data) {
-		key_version = 0;
-	}
 
 	bpage->key_version = key_version;
 
@@ -6402,6 +6410,17 @@ buf_page_decrypt_after_read(
 		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
+	also for pages first compressed and then encrypted. */
+	if (!crypt_data) {
+		key_version = 0;
+	}
+
 	if (page_compressed) {
 		/* the page we read is unencrypted */
 		/* Find free slot from temporary memory array */
@@ -6443,15 +6462,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
@@ -6482,5 +6501,6 @@ buf_page_decrypt_after_read(
 		}
 	}
 
+	fil_space_release(space);
 	return (success);
 }