Merge branch 'bb-10.1-jan-encryption' into bb-10.1-serg

With changes:

* update tests to pass (new encryption/encryption_key_id syntax).
* not merged the code that makes engine aware of the encryption mode
  (CRYPT_SCHEME_1_CBC, CRYPT_SCHEME_1_CTR, storing it on disk, etc),
  because now the encryption plugin is handling it.
* compression+encryption did not work in either branch before the
  merge - and it does not work after the merge. it might be more
  broken after the merge though - some of that code was not merged.
* page checksumming code was not moved (moving of page checksumming
  from fil_space_encrypt() to fil_space_decrypt was not merged).
* restored deleted lines in buf_page_get_frame(), otherwise
  innodb_scrub test failed.

4 files changed, 207 insertions, 211 deletions

diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 58e818ec72a..d3d7b161fa3 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -40,6 +40,7 @@ Created 11/5/1995 Heikki Tuuri
 #include "mem0mem.h"
 #include "btr0btr.h"
 #include "fil0fil.h"
+#include "fil0crypt.h"
 #ifndef UNIV_HOTBACKUP
 #include "buf0buddy.h"
 #include "lock0lock.h"
@@ -54,11 +55,18 @@ Created 11/5/1995 Heikki Tuuri
 #include "page0zip.h"
 #include "srv0mon.h"
 #include "buf0checksum.h"
-#include "fil0pageencryption.h"
 #include "fil0pagecompress.h"
 #include "ut0byte.h"
 #include <new>
 
+#ifdef HAVE_LZO
+#include "lzo/lzo1x.h"
+#endif
+
+/* Number of temporary slots used for encryption/compression
+memory allocation before/after I/O operations */
+#define BUF_MAX_TMP_SLOTS 200
+
 /*
 		IMPLEMENTATION OF THE BUFFER POOL
 		=================================
@@ -504,7 +512,7 @@ buf_page_is_corrupted(
 	ulint		zip_size)	/*!< in: size of compressed page;
 					0 for uncompressed pages */
 {
-	ulint		page_encrypted = fil_page_is_compressed_encrypted(read_buf) || fil_page_is_encrypted(read_buf);
+	ulint		page_encrypted = fil_page_is_encrypted(read_buf);
 	ulint		checksum_field1;
 	ulint		checksum_field2;
 	ibool		crc32_inited = FALSE;
@@ -1001,14 +1009,11 @@ buf_block_init(
 	block->page.state = BUF_BLOCK_NOT_USED;
 	block->page.buf_fix_count = 0;
 	block->page.io_fix = BUF_IO_NONE;
-	block->page.crypt_buf = NULL;
-	block->page.crypt_buf_free = NULL;
-	block->page.comp_buf = NULL;
-	block->page.comp_buf_free = NULL;
 	block->page.key_version = 0;
-	block->page.encrypt_later = false;
-
+	block->page.real_size = 0;
+	block->page.write_size = 0;
 	block->modify_clock = 0;
+	block->page.slot = NULL;
 
 #if defined UNIV_DEBUG_FILE_ACCESSES || defined UNIV_DEBUG
 	block->page.file_page_was_freed = FALSE;
@@ -1361,6 +1366,11 @@ buf_pool_init_instance(
 	/* Initialize the iterator for single page scan search */
 	new(&buf_pool->single_scan_itr) LRUItr(buf_pool, &buf_pool->mutex);
 
+	/* Initialize the temporal memory array and slots */
+	buf_pool->tmp_arr = (buf_tmp_array_t *)mem_zalloc(sizeof(buf_tmp_array_t));
+	buf_pool->tmp_arr->n_slots = BUF_MAX_TMP_SLOTS;
+	buf_pool->tmp_arr->slots = (buf_tmp_buffer_t*)mem_zalloc(sizeof(buf_tmp_buffer_t) * BUF_MAX_TMP_SLOTS);
+
 	buf_pool_mutex_exit(buf_pool);
 
 	return(DB_SUCCESS);
@@ -1412,6 +1422,25 @@ buf_pool_free_instance(
 	ha_clear(buf_pool->page_hash);
 	hash_table_free(buf_pool->page_hash);
 	hash_table_free(buf_pool->zip_hash);
+
+	/* Free all used temporary slots */
+	for(ulint i = 0; i < buf_pool->tmp_arr->n_slots; i++) {
+		buf_tmp_buffer_t* slot = &buf_pool->tmp_arr->slots[i];
+#ifdef HAVE_LZO
+		if (slot->lzo_mem) {
+			ut_free(slot->lzo_mem);
+		}
+#endif
+		if (slot->crypt_buf_free) {
+			ut_free(slot->crypt_buf_free);
+		}
+		if (slot->comp_buf_free) {
+			ut_free(slot->comp_buf_free);
+		}
+	}
+
+	mem_free(buf_pool->tmp_arr->slots);
+	mem_free(buf_pool->tmp_arr);
 }
 
 /********************************************************************//**
@@ -3494,12 +3523,9 @@ buf_page_init_low(
 	bpage->newest_modification = 0;
 	bpage->oldest_modification = 0;
 	bpage->write_size = 0;
-	bpage->crypt_buf = NULL;
-	bpage->crypt_buf_free = NULL;
-	bpage->comp_buf = NULL;
-	bpage->comp_buf_free = NULL;
 	bpage->key_version = 0;
-	bpage->encrypt_later = false;
+	bpage->real_size = 0;
+	bpage->slot = NULL;
 
 	HASH_INVALIDATE(bpage, hash);
 #if defined UNIV_DEBUG_FILE_ACCESSES || defined UNIV_DEBUG
@@ -4366,7 +4392,11 @@ corrupt:
 		}
 	} else {
 		/* io_type == BUF_IO_WRITE */
-		buf_page_encrypt_after_write(bpage);
+		if (bpage->slot) {
+			/* Mark slot free */
+			bpage->slot->reserved = false;
+			bpage->slot = NULL;
+		}
 	}
 
 	buf_pool_mutex_enter(buf_pool);
@@ -5603,17 +5633,78 @@ buf_page_init_for_backup_restore(
 #endif /* !UNIV_HOTBACKUP */
 
 /********************************************************************//**
+Reserve unused slot from temporary memory array and allocate necessary
+temporary memory if not yet allocated.
+@return reserved slot */
+buf_tmp_buffer_t*
+buf_pool_reserve_tmp_slot(
+/*======================*/
+	buf_pool_t*	buf_pool,	/*!< in: buffer pool where to
+					reserve */
+	bool		compressed)	/*!< in: is file space compressed */
+{
+	buf_tmp_buffer_t *free_slot=NULL;
+
+	/* Array is protected by buf_pool mutex */
+	buf_pool_mutex_enter(buf_pool);
+
+	for(ulint i = 0; i < buf_pool->tmp_arr->n_slots; i++) {
+		buf_tmp_buffer_t *slot = &buf_pool->tmp_arr->slots[i];
+
+		if(slot->reserved == false) {
+			free_slot = slot;
+			break;
+		}
+	}
+
+	/* We assume that free slot is found */
+	ut_a(free_slot != NULL);
+	free_slot->reserved = true;
+	/* Now that we have reserved this slot we can release
+	buf_pool mutex */
+	buf_pool_mutex_exit(buf_pool);
+
+	/* Allocate temporary memory for encryption/decryption */
+	if (free_slot->crypt_buf_free == NULL) {
+		free_slot->crypt_buf_free = static_cast<byte *>(ut_malloc(UNIV_PAGE_SIZE*2));
+		free_slot->crypt_buf = static_cast<byte *>(ut_align(free_slot->crypt_buf_free, UNIV_PAGE_SIZE));
+		memset(free_slot->crypt_buf_free, 0, UNIV_PAGE_SIZE *2);
+	}
+
+	/* For page compressed tables allocate temporary memory for
+	compression/decompression */
+	if (compressed && free_slot->comp_buf_free == NULL) {
+		free_slot->comp_buf_free = static_cast<byte *>(ut_malloc(UNIV_PAGE_SIZE*2));
+		free_slot->comp_buf = static_cast<byte *>(ut_align(free_slot->comp_buf_free, UNIV_PAGE_SIZE));
+		memset(free_slot->comp_buf_free, 0, UNIV_PAGE_SIZE *2);
+#ifdef HAVE_LZO
+		free_slot->lzo_mem = static_cast<byte *>(ut_malloc(LZO1X_1_15_MEM_COMPRESS));
+		memset(free_slot->lzo_mem, 0, LZO1X_1_15_MEM_COMPRESS);
+#endif
+	}
+
+	return (free_slot);
+}
+
+/********************************************************************//**
 Encrypts a buffer page right before it's flushed to disk
 */
 byte*
 buf_page_encrypt_before_write(
 /*==========================*/
 	buf_page_t* bpage,     /*!< in/out: buffer page to be flushed */
-	const byte* src_frame) /*!< in: src frame */
+	const byte* src_frame, /*!< in: src frame */
+	ulint space_id) /*!< in: space id */
 {
-	bpage->encrypt_later = false;
+	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space_id);
+	bpage->real_size = UNIV_PAGE_SIZE;
 
-	if (srv_encrypt_tables == FALSE) {
+	if (crypt_data != NULL && crypt_data->encryption == FIL_SPACE_ENCRYPTION_OFF) {
+		/* Encryption is disabled */
+		return const_cast<byte*>(src_frame);
+	}
+
+	if (!srv_encrypt_tables && (crypt_data == NULL || crypt_data->encryption == FIL_SPACE_ENCRYPTION_DEFAULT)) {
 		/* Encryption is disabled */
 		return const_cast<byte*>(src_frame);
 	}
@@ -5638,101 +5729,61 @@ buf_page_encrypt_before_write(
 
 	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);
+
+	/* Find free slot from temporary memory array */
+	buf_tmp_buffer_t* slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
+	bpage->slot = slot;
 
-	/**
-	* TODO(jonaso): figure out more clever malloc strategy
-	*
-	* This implementation does a malloc/free per iop for encrypted
-	* tablespaces. Alternative strategies that have been considered are
-	*
-	* 1) use buf_block_alloc (i.e alloc from buffer pool)
-	*    this does not work as buf_block_alloc will then be called
-	*    when needing to flush a page, which might be triggered
-	*    due to shortage of memory in buffer pool
-	* 2) allocate a buffer per fil_node_t
-	*    this would break abstraction layers and has therfore not been
-	*    considered a lot.
-	*/
-
-	if (bpage->crypt_buf_free == NULL) {
-		bpage->crypt_buf_free = (byte*)malloc(page_size*2);
-		// TODO: Is 4k aligment enough ?
-		bpage->crypt_buf = (byte *)ut_align(bpage->crypt_buf_free, page_size);
-	}
-
-	byte *dst_frame = bpage->crypt_buf;
-
-	if (!fil_space_is_page_compressed(bpage->space)) {
-		// encrypt page content
-		fil_space_encrypt(bpage->space, bpage->offset,
-			bpage->newest_modification,
-			src_frame, zip_size, dst_frame, 0);
+	byte *dst_frame = slot->crypt_buf;
+
+	if (!page_compressed) {
+		/* Encrypt page content */
+		fil_space_encrypt(bpage->space,
+				bpage->offset,
+				bpage->newest_modification,
+				src_frame,
+				zip_size,
+				dst_frame);
 
 		unsigned key_version =
 			mach_read_from_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
 		ut_ad(key_version == 0 || key_version >= bpage->key_version);
 		bpage->key_version = key_version;
+		bpage->real_size = page_size;
 	} else {
-		/** Compression and encryption is done later at os0file.cc */
-		bpage->encrypt_later = true;
-		dst_frame = (byte *)src_frame;
-	}
-
-	// return dst_frame which will be written
-	return dst_frame;
-}
+		/* First we compress the page content */
+		ulint out_len = 0;
+		ulint block_size = fil_space_get_block_size(bpage->space, bpage->offset, page_size);
+
+		byte *tmp = fil_compress_page(bpage->space,
+					(byte *)src_frame,
+					slot->comp_buf,
+					page_size,
+					fil_space_get_page_compression_level(bpage->space),
+					block_size,
+					&out_len,
+#ifdef HAVE_LZO
+					slot->lzo_mem
+#else
+					NULL
+#endif
+					);
 
-/********************************************************************//**
-Release memory after encrypted page has been written to disk
-*/
-ibool
-buf_page_encrypt_after_write(
-/*=========================*/
-	buf_page_t* bpage) /*!< in/out: buffer page flushed */
-{
-	if (bpage->crypt_buf_free != NULL) {
-		free(bpage->crypt_buf_free);
-		bpage->crypt_buf_free = NULL;
-		bpage->crypt_buf = NULL;
-	}
+		bpage->real_size = out_len;
 
-	if (bpage->comp_buf_free != NULL) {
-		free(bpage->comp_buf_free);
-		bpage->comp_buf_free = NULL;
-		bpage->comp_buf = NULL;
+		/* And then we encrypt the page content */
+		fil_space_encrypt(bpage->space,
+				bpage->offset,
+				bpage->newest_modification,
+		        	tmp,
+				zip_size,
+				dst_frame);
 	}
 
-	return (TRUE);
-}
-
-/********************************************************************//**
-Allocates memory to read in an encrypted page
-*/
-byte*
-buf_page_decrypt_before_read(
-/*=========================*/
-        buf_page_t* bpage, /*!< in/out: buffer page to be read */
-	ulint	zip_size)  /*!< in: compressed page size, or 0 */
-{
-	ulint size = (zip_size) ? zip_size : UNIV_PAGE_SIZE;
-
-        /*
-          Here we only need to allocate space for not header pages
-          in case of file space encryption.  Table encryption is handled
-          later.
-        */
-	if (!srv_encrypt_tables || bpage->offset == 0 ||
-            fil_space_check_encryption_read(bpage->space) == false)
-          return zip_size ? bpage->zip.data : ((buf_block_t*) bpage)->frame;
-
-        if (bpage->crypt_buf_free == NULL)
-        {
-          // allocate buffer to read data into
-          bpage->crypt_buf_free = (byte*)malloc(size*2);
-          // TODO: Is 4K aligment enough ?
-          bpage->crypt_buf = (byte*)ut_align(bpage->crypt_buf_free, size);
-        }
-        return bpage->crypt_buf;
+	// return dst_frame which will be written
+	return dst_frame;
 }
 
 /********************************************************************//**
@@ -5747,7 +5798,6 @@ buf_page_decrypt_after_read(
 
 	if (bpage->offset == 0) {
 		/* File header pages are not encrypted */
-		ut_a(bpage->crypt_buf == NULL);
 		return (TRUE);
 	}
 
@@ -5756,91 +5806,55 @@ buf_page_decrypt_after_read(
 
 	byte* dst_frame = (zip_size) ? bpage->zip.data :
 		((buf_block_t*) bpage)->frame;
-	const byte* src_frame = bpage->crypt_buf != NULL ?
-		bpage->crypt_buf : dst_frame;
-
 	unsigned key_version =
-		mach_read_from_4(src_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
+		mach_read_from_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
+
+	bool page_compressed = fil_page_is_compressed(dst_frame);
 
-	bool page_compressed_encrypted = fil_page_is_compressed_encrypted(dst_frame);
+	buf_pool_t* buf_pool = buf_pool_from_bpage(bpage);
 
 	if (key_version == 0) {
 		/* the page we read is unencrypted */
 
-		if (fil_page_is_compressed(dst_frame)) {
-			if (bpage->comp_buf_free == NULL) {
-				bpage->comp_buf_free = (byte *)malloc(UNIV_PAGE_SIZE*2);
-				// TODO: is 4k aligment enough ?
-				bpage->comp_buf = (byte*)ut_align(bpage->comp_buf_free, UNIV_PAGE_SIZE);
-			}
+		if (page_compressed) {
+			/* Find free slot from temporary memory array */
+			buf_tmp_buffer_t* slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
 
-			fil_decompress_page(bpage->comp_buf, dst_frame, size, NULL);
-		} else {
-			if (dst_frame != src_frame) {
-				/* but we had allocated a crypt_buf */
-				// TODO: Can this be avoided ?
-				memcpy(dst_frame, src_frame, size);
-			}
+			fil_decompress_page(slot->comp_buf,
+					dst_frame,
+					size,
+					&bpage->write_size);
+
+			/* Mark this slot as free */
+			slot->reserved = false;
 		}
 	} else {
 		/* the page we read is encrypted */
-		if (dst_frame == src_frame) {
-			/* but we had NOT allocated a crypt buf
-			* malloc a buffer, copy page to it
-			* and then decrypt from that into real page*/
-			bpage->crypt_buf_free = (byte *)malloc(UNIV_PAGE_SIZE*2);
-			// TODO: is 4k aligment enough ?
-			src_frame = bpage->crypt_buf = (byte*)ut_align(bpage->crypt_buf_free, UNIV_PAGE_SIZE);
-			memcpy(bpage->crypt_buf, dst_frame, size);
-		}
+		/* Find free slot from temporary memory array */
+		buf_tmp_buffer_t* slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
+
+		memcpy(slot->crypt_buf, dst_frame, size);
 
-		/* decrypt from src_frame to dst_frame */
+		/* decrypt from crypt_buf to dst_frame */
 		fil_space_decrypt(bpage->space,
-				  src_frame, size, dst_frame);
+				  slot->crypt_buf,
+				  size,
+				  dst_frame);
 
 		/* decompress from dst_frame to comp_buf and then copy to
 		buffer pool */
-		if (page_compressed_encrypted) {
-			if (bpage->comp_buf_free == NULL) {
-				bpage->comp_buf_free = (byte *)malloc(UNIV_PAGE_SIZE*2);
-				// TODO: is 4k aligment enough ?
-				bpage->comp_buf = (byte*)ut_align(bpage->comp_buf_free, UNIV_PAGE_SIZE);
-			}
-
-			fil_decompress_page(bpage->comp_buf, dst_frame, size, NULL);
+		if (page_compressed) {
+			fil_decompress_page(slot->comp_buf,
+					dst_frame,
+					size,
+					&bpage->write_size);
 		}
-	}
-
-	bpage->key_version = key_version;
 
-	if (bpage->crypt_buf_free != NULL) {
-		// free temp page
-		free(bpage->crypt_buf_free);
-		bpage->crypt_buf = NULL;
-		bpage->crypt_buf_free = NULL;
+		/* Mark this slot as free */
+		slot->reserved = false;
 	}
 
-	if (bpage->comp_buf_free != NULL) {
-		// free temp page
-		free(bpage->comp_buf_free);
-		bpage->comp_buf = NULL;
-		bpage->comp_buf_free = NULL;
-	}
+	bpage->key_version = key_version;
 
 	return (TRUE);
 }
-
-/********************************************************************//**
-Release memory allocated for decryption
-*/
-void
-buf_page_decrypt_cleanup(
-/*=====================*/
-	buf_page_t* bpage) /*!< in/out: buffer page */
-{
-	if (bpage->crypt_buf != NULL) {
-		free(bpage->crypt_buf_free);
-		bpage->crypt_buf = NULL;
-		bpage->crypt_buf_free = NULL;
-	}
-}
diff --git a/storage/innobase/buf/buf0dblwr.cc b/storage/innobase/buf/buf0dblwr.cc
index ccc39e499ca..15a46abe55f 100644
--- a/storage/innobase/buf/buf0dblwr.cc
+++ b/storage/innobase/buf/buf0dblwr.cc
@@ -36,6 +36,7 @@ Created 2011/12/19
 #include "srv0srv.h"
 #include "page0zip.h"
 #include "trx0sys.h"
+#include "fil0crypt.h"
 
 #ifndef UNIV_HOTBACKUP
 
@@ -385,7 +386,7 @@ buf_dblwr_init_or_load_pages(
 	/* Read the trx sys header to check if we are using the doublewrite
 	buffer */
 	off_t  trx_sys_page = TRX_SYS_PAGE_NO * UNIV_PAGE_SIZE;
-	os_file_read(file, read_buf, trx_sys_page, UNIV_PAGE_SIZE, FALSE);
+	os_file_read(file, read_buf, trx_sys_page, UNIV_PAGE_SIZE);
 
 	doublewrite = read_buf + TRX_SYS_DOUBLEWRITE;
 
@@ -430,9 +431,9 @@ buf_dblwr_init_or_load_pages(
 
         block_bytes = TRX_SYS_DOUBLEWRITE_BLOCK_SIZE * UNIV_PAGE_SIZE;
 
-	os_file_read(file, buf, block1 * UNIV_PAGE_SIZE, block_bytes, FALSE);
+	os_file_read(file, buf, block1 * UNIV_PAGE_SIZE, block_bytes);
 	os_file_read(file, buf + block_bytes, block2 * UNIV_PAGE_SIZE,
-		     block_bytes, FALSE);
+		     block_bytes);
 
 	/* Check if any of these pages is half-written in data files, in the
 	intended position */
@@ -530,9 +531,7 @@ buf_dblwr_process()
 				zip_size ? zip_size : UNIV_PAGE_SIZE,
 				read_buf,
 				NULL,
-				0,
-				0,
-				false);
+				0);
 
 			if (fil_space_verify_crypt_checksum(read_buf, zip_size)) {
 				/* page is encrypted and checksum is OK */
@@ -593,9 +592,7 @@ buf_dblwr_process()
 					zip_size ? zip_size : UNIV_PAGE_SIZE,
 					page,
 					NULL,
-					0,
-					0,
-					false);
+					0);
 
 				ib_logf(IB_LOG_LEVEL_INFO,
 					"Recovered the page from"
@@ -620,9 +617,7 @@ buf_dblwr_process()
 						zip_size ? zip_size : UNIV_PAGE_SIZE,
 						page,
 						NULL,
-						0,
-						0,
-						false);
+						0);
 				}
 			}
 		}
@@ -640,9 +635,9 @@ buf_dblwr_process()
 		memset(buf, 0, bytes);
 
 		fil_io(OS_FILE_WRITE, true, TRX_SYS_SPACE, 0,
-			buf_dblwr->block1, 0, bytes, buf, NULL, NULL, 0, false);
+			buf_dblwr->block1, 0, bytes, buf, NULL, NULL);
 		fil_io(OS_FILE_WRITE, true, TRX_SYS_SPACE, 0,
-			buf_dblwr->block2, 0, bytes, buf, NULL, NULL, 0, false);
+			buf_dblwr->block2, 0, bytes, buf, NULL, NULL);
 
 		ut_free(unaligned_buf);
         }
@@ -856,9 +851,7 @@ buf_dblwr_write_block_to_datafile(
 			buf_page_get_zip_size(bpage),
 			frame,
 			(void*) bpage,
-			0,
-			bpage->newest_modification,
-			bpage->encrypt_later);
+			0);
 
 		return;
 	}
@@ -874,12 +867,10 @@ buf_dblwr_write_block_to_datafile(
 		0,
 		buf_block_get_page_no(block),
 		0,
-		UNIV_PAGE_SIZE,
+		bpage->real_size,
 		frame,
 		(void*) block,
-		(ulint *)&bpage->write_size,
-		bpage->newest_modification,
-		bpage->encrypt_later);
+		(ulint *)&bpage->write_size);
 }
 
 /********************************************************************//**
@@ -973,7 +964,7 @@ try_again:
 
 	fil_io(OS_FILE_WRITE, true, TRX_SYS_SPACE, 0,
 	       buf_dblwr->block1, 0, len,
-		(void*) write_buf, NULL, 0, 0, false);
+		(void*) write_buf, NULL, 0);
 
 	if (buf_dblwr->first_free <= TRX_SYS_DOUBLEWRITE_BLOCK_SIZE) {
 		/* No unwritten pages in the second block. */
@@ -989,7 +980,7 @@ try_again:
 
 	fil_io(OS_FILE_WRITE, true, TRX_SYS_SPACE, 0,
 	       buf_dblwr->block2, 0, len,
-		(void*) write_buf, NULL, 0, 0, false);
+		(void*) write_buf, NULL, 0);
 
 flush:
 	/* increment the doublewrite flushed pages counter */
@@ -1226,9 +1217,7 @@ retry:
 			UNIV_PAGE_SIZE,
 			(void*) (buf_dblwr->write_buf + UNIV_PAGE_SIZE * i),
 			NULL,
-			0,
-			bpage->newest_modification,
-			bpage->encrypt_later);
+			0);
 	} else {
 		/* It is a regular page. Write it directly to the
 		doublewrite buffer */
@@ -1237,12 +1226,10 @@ retry:
 			TRX_SYS_SPACE, 0,
 			offset,
 			0,
-			UNIV_PAGE_SIZE,
+			bpage->real_size,
 			frame,
 			NULL,
-			0,
-			bpage->newest_modification,
-			bpage->encrypt_later);
+			0);
 	}
 
 	/* Now flush the doublewrite buffer data to disk */
diff --git a/storage/innobase/buf/buf0flu.cc b/storage/innobase/buf/buf0flu.cc
index 9bd7a7c007d..d893d424b02 100644
--- a/storage/innobase/buf/buf0flu.cc
+++ b/storage/innobase/buf/buf0flu.cc
@@ -909,7 +909,7 @@ buf_flush_write_block_low(
 		break;
 	}
 
-	frame = buf_page_encrypt_before_write(bpage, frame);
+	frame = buf_page_encrypt_before_write(bpage, frame, space_id);
 
 	if (!srv_use_doublewrite_buf || !buf_dblwr) {
 		fil_io(OS_FILE_WRITE | OS_AIO_SIMULATED_WAKE_LATER,
@@ -918,12 +918,10 @@ buf_flush_write_block_low(
 			zip_size,
 			buf_page_get_page_no(bpage),
 			0,
-			zip_size ? zip_size : UNIV_PAGE_SIZE,
+			zip_size ? zip_size : bpage->real_size,
 			frame,
 			bpage,
-			&bpage->write_size,
-			bpage->newest_modification,
-			bpage->encrypt_later);
+			&bpage->write_size);
 	} else {
 
 		/* InnoDB uses doublewrite buffer and doublewrite buffer
@@ -940,12 +938,10 @@ buf_flush_write_block_low(
 				zip_size,
 				buf_page_get_page_no(bpage),
 				0,
-				zip_size ? zip_size : UNIV_PAGE_SIZE,
+				zip_size ? zip_size : bpage->real_size,
 				frame,
 				bpage,
-				&bpage->write_size,
-				bpage->newest_modification,
-				bpage->encrypt_later);
+				&bpage->write_size);
 		} else if (flush_type == BUF_FLUSH_SINGLE_PAGE) {
 			buf_dblwr_write_single_page(bpage, sync);
 		} else {
diff --git a/storage/innobase/buf/buf0rea.cc b/storage/innobase/buf/buf0rea.cc
index e91a5da6621..b4b474c547f 100644
--- a/storage/innobase/buf/buf0rea.cc
+++ b/storage/innobase/buf/buf0rea.cc
@@ -176,7 +176,7 @@ buf_read_page_low(
 
 	ut_ad(buf_page_in_file(bpage));
 
-	byte* frame = buf_page_decrypt_before_read(bpage, zip_size);
+	byte* frame = zip_size ? bpage->zip.data : ((buf_block_t*) bpage)->frame;
 
 	if (sync) {
 		thd_wait_begin(NULL, THD_WAIT_DISKIO);
@@ -186,7 +186,7 @@ buf_read_page_low(
 		*err = fil_io(OS_FILE_READ | wake_later
 			| ignore_nonexistent_pages,
 			sync, space, zip_size, offset, 0, zip_size,
-			frame, bpage, &bpage->write_size, 0, false);
+			frame, bpage, &bpage->write_size);
 	} else {
 		ut_a(buf_page_get_state(bpage) == BUF_BLOCK_FILE_PAGE);
 
@@ -194,7 +194,7 @@ buf_read_page_low(
 			| ignore_nonexistent_pages,
 			sync, space, 0, offset, 0, UNIV_PAGE_SIZE,
 			frame, bpage,
-			&bpage->write_size, 0, false);
+			&bpage->write_size);
 	}
 
 	if (sync) {
@@ -202,7 +202,6 @@ buf_read_page_low(
 	}
 
 	if (*err != DB_SUCCESS) {
-		buf_page_decrypt_cleanup(bpage);
 		if (ignore_nonexistent_pages || *err == DB_TABLESPACE_DELETED) {
 			buf_read_page_handle_error(bpage);
 			return(0);