3 files changed, 78 insertions, 75 deletions

diff --git a/storage/xtradb/buf/buf0buf.cc b/storage/xtradb/buf/buf0buf.cc
index ebf6bb10caa..c57dab79ef7 100644
--- a/storage/xtradb/buf/buf0buf.cc
+++ b/storage/xtradb/buf/buf0buf.cc
@@ -65,6 +65,15 @@ Created 11/5/1995 Heikki Tuuri
 #include "fil0pagecompress.h"
 #include "ha_prototypes.h"
 
+/** Decrypt a page.
+@param[in,out]	bpage	Page control block
+@param[in,out]	space	tablespace
+@return whether the operation was successful */
+static
+bool
+buf_page_decrypt_after_read(buf_page_t* bpage, fil_space_t* space)
+	MY_ATTRIBUTE((nonnull));
+
 /* prototypes for new functions added to ha_innodb.cc */
 trx_t* innobase_get_trx();
 
@@ -548,16 +557,13 @@ buf_block_alloc(
 }
 #endif /* !UNIV_HOTBACKUP */
 
-/********************************************************************//**
-Checks if a page is all zeroes.
-@return	TRUE if the page is all zeroes */
+/** Check if a page is all zeroes.
+@param[in]	read_buf	database page
+@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@return	whether the page is all zeroes */
 UNIV_INTERN
 bool
-buf_page_is_zeroes(
-/*===============*/
-	const byte*	read_buf,	/*!< in: a database page */
-	const ulint	zip_size)	/*!< in: size of compressed page;
-					0 for uncompressed pages */
+buf_page_is_zeroes(const byte* read_buf, ulint zip_size)
 {
 	const ulint page_size = zip_size ? zip_size : UNIV_PAGE_SIZE;
 
@@ -673,8 +679,7 @@ buf_page_is_checksum_valid_none(
 	       && checksum_field1 == BUF_NO_CHECKSUM_MAGIC);
 }
 
-/********************************************************************//**
-Checks if a page is corrupt.
+/** Check if a page is corrupt.
 @param[in]	check_lsn		true if LSN should be checked
 @param[in]	read_buf		Page to be checked
 @param[in]	zip_size		compressed size or 0
@@ -4529,34 +4534,30 @@ buf_mark_space_corrupt(
 	mutex_exit(&buf_pool->LRU_list_mutex);
 }
 
-/********************************************************************//**
-Check if page is maybe compressed, encrypted or both when we encounter
+/** Check if page is maybe compressed, encrypted or both when we encounter
 corrupted page. Note that we can't be 100% sure if page is corrupted
 or decrypt/decompress just failed.
-@param[in,out]	bpage		Page
-@return DB_SUCCESS if page has been read and is not corrupted,
-@retval DB_PAGE_CORRUPTED if page based on checksum check is corrupted,
-@retval DB_DECRYPTION_FAILED if page post encryption checksum matches but
+@param[in,out]	bpage		page
+@param[in,out]	space		tablespace from fil_space_acquire_for_io()
+@return	whether the operation succeeded
+@retval	DB_SUCCESS		if page has been read and is not corrupted
+@retval	DB_PAGE_CORRUPTED	if page based on checksum check is corrupted
+@retval	DB_DECRYPTION_FAILED	if page post encryption checksum matches but
 after decryption normal page checksum does not match.
-@retval DB_TABLESPACE_DELETED if accessed tablespace is not found */
+@retval	DB_TABLESPACE_DELETED	if accessed tablespace is not found */
 static
 dberr_t
-buf_page_check_corrupt(buf_page_t*	bpage)
+buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 {
+	ut_ad(space->n_pending_ios > 0);
+
 	ulint zip_size = buf_page_get_zip_size(bpage);
 	byte* dst_frame = (zip_size) ? bpage->zip.data :
 		((buf_block_t*) bpage)->frame;
-	FilSpace space(bpage->space, true);
 	bool still_encrypted = false;
 	dberr_t err = DB_SUCCESS;
 	bool corrupted = false;
-	fil_space_crypt_t* crypt_data = NULL;
-
-	if (!space()) {
-		return(DB_TABLESPACE_DELETED);
-	}
-
-	crypt_data = space()->crypt_data;
+	fil_space_crypt_t* 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
@@ -4568,12 +4569,12 @@ buf_page_check_corrupt(buf_page_t*	bpage)
 		crypt_data->type != CRYPT_SCHEME_UNENCRYPTED &&
 		!bpage->encrypted &&
 		fil_space_verify_crypt_checksum(dst_frame, zip_size,
-			space(), bpage->offset));
-
+			space, bpage->offset));
 	if (!still_encrypted) {
 		/* If traditional checksums match, we assume that page is
 		not anymore encrypted. */
-		corrupted = buf_page_is_corrupted(true, dst_frame, zip_size, space());
+		corrupted = buf_page_is_corrupted(true, dst_frame, zip_size,
+						  space);
 
 		if (!corrupted) {
 			bpage->encrypted = false;
@@ -4596,7 +4597,7 @@ buf_page_check_corrupt(buf_page_t*	bpage)
 			", page number=%u]"
 			" in file %s cannot be decrypted.",
 			bpage->space, bpage->offset,
-			space()->name);
+			space->name);
 
 		ib_logf(IB_LOG_LEVEL_INFO,
 			"However key management plugin or used key_version " ULINTPF
@@ -4614,26 +4615,23 @@ buf_page_check_corrupt(buf_page_t*	bpage)
 	return (err);
 }
 
-/********************************************************************//**
-Completes an asynchronous read or write request of a file page to or from
-the buffer pool.
+/** Complete a read or write request of a file page to or from the buffer pool.
 @param[in,out]		bpage		Page to complete
-@return DB_SUCCESS if page has been read and is not corrupted,
-DB_PAGE_CORRUPTED if page based on checksum check is corrupted,
-DB_DECRYPTION_FAILED if page post encryption checksum matches but
-after decryption normal page checksum does not match.
-in write only DB_SUCCESS is possible. */
+@return whether the operation succeeded
+@retval	DB_SUCCESS		always when writing, or if a read page was OK
+@retval	DB_PAGE_CORRUPTED	if the checksum fails on a page read
+@retval	DB_DECRYPTION_FAILED	if page post encryption checksum matches but
+				after decryption normal page checksum does
+				not match */
 UNIV_INTERN
 dberr_t
-buf_page_io_complete(
-	buf_page_t*	bpage)
+buf_page_io_complete(buf_page_t* bpage)
 {
 	enum buf_io_fix	io_type;
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 	const ibool	uncompressed = (buf_page_get_state(bpage)
 					== BUF_BLOCK_FILE_PAGE);
 	bool		have_LRU_mutex = false;
-	fil_space_t*	space = NULL;
 	byte*		frame = NULL;
 	dberr_t 	err = DB_SUCCESS;
 
@@ -4653,7 +4651,13 @@ buf_page_io_complete(
 		ulint	read_space_id = 0;
 		uint	key_version = 0;
 
-		buf_page_decrypt_after_read(bpage);
+		ut_ad(bpage->zip.data || ((buf_block_t*)bpage)->frame);
+		fil_space_t* space = fil_space_acquire_for_io(bpage->space);
+		if (!space) {
+			return(DB_TABLESPACE_DELETED);
+		}
+
+		buf_page_decrypt_after_read(bpage, space);
 
 		if (buf_page_get_zip_size(bpage)) {
 			frame = bpage->zip.data;
@@ -4727,7 +4731,7 @@ buf_page_io_complete(
 
 		if (UNIV_LIKELY(!bpage->is_corrupt ||
 				!srv_pass_corrupt_table)) {
-			err = buf_page_check_corrupt(bpage);
+			err = buf_page_check_corrupt(bpage, space);
 		}
 
 database_corrupted:
@@ -4740,6 +4744,7 @@ database_corrupted:
 					buf_mark_space_corrupt(bpage);
 					ib_logf(IB_LOG_LEVEL_INFO,
 						"Simulated page corruption");
+					fil_space_release_for_io(space);
 					return(err);
 				}
 				err = DB_SUCCESS;
@@ -4747,9 +4752,6 @@ database_corrupted:
 			);
 
 			if (err == DB_PAGE_CORRUPTED) {
-				fil_system_enter();
-				space = fil_space_get_by_id(bpage->space);
-
 				ib_logf(IB_LOG_LEVEL_ERROR,
 					"Database page corruption on disk"
 					" or a failed file read of tablespace %s"
@@ -4760,8 +4762,6 @@ database_corrupted:
 					space->name,
 					bpage->space, bpage->offset);
 
-				fil_system_exit();
-
 				buf_page_print(frame, buf_page_get_zip_size(bpage),
 					BUF_PAGE_PRINT_NO_CRASH);
 
@@ -4798,6 +4798,7 @@ database_corrupted:
 				table as corrupted instead of crashing server */
 				if (bpage->space > TRX_SYS_SPACE) {
 					buf_mark_space_corrupt(bpage);
+					fil_space_release_for_io(space);
 					return(err);
 				} else {
 					ib_logf(IB_LOG_LEVEL_FATAL,
@@ -4836,6 +4837,8 @@ database_corrupted:
 			}
 
 		}
+
+		fil_space_release_for_io(space);
 	} else {
 		/* io_type == BUF_IO_WRITE */
 		if (bpage->slot) {
@@ -6306,16 +6309,17 @@ buf_page_encrypt_before_write(
 	return dst_frame;
 }
 
-/********************************************************************//**
-Decrypt page after it has been read from disk
-@param[in,out]	bpage		Page control block
-@return true if successfull, false if something went wrong
-*/
-UNIV_INTERN
+/** Decrypt a page.
+@param[in,out]	bpage	Page control block
+@param[in,out]	space	tablespace
+@return whether the operation was successful */
+static
 bool
-buf_page_decrypt_after_read(
-	buf_page_t*	bpage)
+buf_page_decrypt_after_read(buf_page_t* bpage, fil_space_t* space)
 {
+	ut_ad(space->n_pending_ios > 0);
+	ut_ad(space->id == bpage->space);
+
 	ulint zip_size = buf_page_get_zip_size(bpage);
 	ulint size = (zip_size) ? zip_size : UNIV_PAGE_SIZE;
 
@@ -6333,12 +6337,10 @@ buf_page_decrypt_after_read(
 		return (true);
 	}
 
-	FilSpace space(bpage->space, false, true);
-
 	/* 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 (!space() || !space()->crypt_data) {
+	if (!space->crypt_data) {
 		key_version = 0;
 	}
 
@@ -6375,8 +6377,8 @@ buf_page_decrypt_after_read(
 
 				/* Mark page encrypted in case it should
 				be. */
-				if (key_version && space()->crypt_data &&
-				    space()->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED) {
+				if (space->crypt_data->type
+				    != CRYPT_SCHEME_UNENCRYPTED) {
 					bpage->encrypted = true;
 				}
 
@@ -6391,12 +6393,8 @@ buf_page_decrypt_after_read(
 #endif
 
 			/* decrypt using crypt_buf to dst_frame */
-			byte* res = fil_space_decrypt(space(),
-						slot->crypt_buf,
-						dst_frame,
-						&bpage->encrypted);
-
-			if (!res) {
+			if (!fil_space_decrypt(space, slot->crypt_buf,
+					       dst_frame, &bpage->encrypted)) {
 				success = false;
 			}
 
@@ -6427,5 +6425,6 @@ buf_page_decrypt_after_read(
 		}
 	}
 
+	ut_ad(space->n_pending_ios > 0);
 	return (success);
 }
diff --git a/storage/xtradb/buf/buf0flu.cc b/storage/xtradb/buf/buf0flu.cc
index 62a2468ba66..1f5c3993be7 100644
--- a/storage/xtradb/buf/buf0flu.cc
+++ b/storage/xtradb/buf/buf0flu.cc
@@ -873,7 +873,7 @@ buf_flush_write_block_low(
 	buf_flush_t	flush_type,	/*!< in: type of flush */
 	bool		sync)		/*!< in: true if sync IO request */
 {
-	fil_space_t*	space = fil_space_acquire(bpage->space, true);
+	fil_space_t*	space = fil_space_acquire_for_io(bpage->space);
 	if (!space) {
 		return;
 	}
@@ -995,6 +995,13 @@ buf_flush_write_block_low(
 		ut_ad(flush_type == BUF_FLUSH_SINGLE_PAGE);
 		fil_flush(space);
 
+		/* The tablespace could already have been dropped,
+		because fil_io(request, sync) would already have
+		decremented the node->n_pending. However,
+		buf_page_io_complete() only needs to look up the
+		tablespace during read requests, not during writes. */
+		ut_ad(buf_page_get_io_fix_unlocked(bpage) == BUF_IO_WRITE);
+
 #ifdef UNIV_DEBUG
 		dberr_t err =
 #endif
@@ -1003,7 +1010,7 @@ buf_flush_write_block_low(
 		ut_ad(err == DB_SUCCESS);
 	}
 
-	fil_space_release(space);
+	fil_space_release_for_io(space);
 
 	/* Increment the counter of I/O operations used
 	for selecting LRU policy. */
@@ -2864,7 +2871,7 @@ DECLARE_THREAD(buf_flush_page_cleaner_thread)(
 		success = buf_flush_list(PCT_IO(100), LSN_MAX, &n_flushed);
 		buf_flush_wait_batch_end(NULL, BUF_FLUSH_LIST);
 
-	} while (!success || n_flushed > 0);
+	} while (!success || n_flushed > 0 || (IS_XTRABACKUP() && buf_get_n_pending_read_ios() > 0));
 
 	/* Some sanity checks */
 	ut_a(srv_get_active_thread_type() == SRV_NONE);
diff --git a/storage/xtradb/buf/buf0rea.cc b/storage/xtradb/buf/buf0rea.cc
index e275eead4cc..85b04d37a08 100644
--- a/storage/xtradb/buf/buf0rea.cc
+++ b/storage/xtradb/buf/buf0rea.cc
@@ -955,11 +955,8 @@ buf_read_ibuf_merge_pages(
 
 tablespace_deleted:
 			/* We have deleted or are deleting the single-table
-			tablespace: remove the entries for that page */
-
-			ibuf_merge_or_delete_for_page(NULL, space_ids[i],
-						      page_nos[i],
-						      zip_size, FALSE);
+			tablespace: remove the entries for tablespace. */
+			ibuf_delete_for_discarded_space(space_ids[i]);
 			break;
 		case DB_DECRYPTION_FAILED:
 			ib_logf(IB_LOG_LEVEL_ERROR,