MDEV-23190 InnoDB data file extension is not crash-safe

When InnoDB is extending a data file, it is updating the FSP_SIZE
field in the first page of the data file.

In commit 8451e090 (MDEV-11556)
we removed a work-around for this bug and made recovery stricter,
by making it track changes to FSP_SIZE via redo log records, and
extend the data files before any changes are being applied to them.

It turns out that the function fsp_fill_free_list() is not crash-safe
with respect to this when it is initializing the change buffer bitmap
page (page 1, or generally, N*innodb_page_size+1). It uses a separate
mini-transaction that is committed (and will be written to the redo
log file) before the mini-transaction that actually extended the data
file. Hence, recovery can observe a reference to a page that is
beyond the current end of the data file.

fsp_fill_free_list(): Initialize the change buffer bitmap page in
the same mini-transaction.

The rest of the changes are fixing a bug that the use of the separate
mini-transaction was attempting to work around. Namely, we must ensure
that no other thread will access the change buffer bitmap page before
our mini-transaction has been committed and all page latches have been
released.

That is, for read-ahead as well as neighbour flushing, we must avoid
accessing pages that might not yet be durably part of the tablespace.

fil_space_t::committed_size: The size of the tablespace
as persisted by mtr_commit().

fil_space_t::max_page_number_for_io(): Limit the highest page
number for I/O batches to committed_size.

MTR_MEMO_SPACE_X_LOCK: Replaces MTR_MEMO_X_LOCK for fil_space_t::latch.

mtr_x_space_lock(): Replaces mtr_x_lock() for fil_space_t::latch.

mtr_memo_slot_release_func(): When releasing MTR_MEMO_SPACE_X_LOCK,
copy space->size to space->committed_size. In this way, read-ahead
or flushing will never be invoked on pages that do not yet exist
according to FSP_SIZE.

storage/innobase/buf/buf0flu.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2019, MariaDB Corporation.
+Copyright (c) 2013, 2020, MariaDB Corporation.
 Copyright (c) 2013, 2014, Fusion-io
 
 This program is free software; you can redistribute it and/or modify it under
@@ -1241,8 +1241,11 @@ buf_flush_try_neighbors(
 	/* fprintf(stderr, "Flush area: low %lu high %lu\n", low, high); */
 #endif
 
-	if (high > fil_space_get_size(space)) {
-		high = fil_space_get_size(space);
+	if (fil_space_t *s = fil_space_acquire_for_io(space)) {
+		high = s->max_page_number_for_io(high);
+		fil_space_release_for_io(s);
+	} else {
+		return 0;
 	}
 
 	ulint	count = 0;

storage/innobase/buf/buf0rea.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2013, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2017, MariaDB Corporation.
+Copyright (c) 2015, 2020, 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
@@ -292,19 +292,22 @@ buf_read_ahead_random(
 		return(0);
 	}
 
-	/* Remember the tablespace version before we ask te tablespace size
-	below: if DISCARD + IMPORT changes the actual .ibd file meanwhile, we
-	do not try to read outside the bounds of the tablespace! */
+	if (fil_space_t *s = fil_space_acquire_for_io(space)) {
+		/* Remember the tablespace version along with the
+		tablespace size: if DISCARD + IMPORT changes the
+		actual .ibd file meanwhile, we do not try to read
+		outside the bounds of the tablespace! */
+		tablespace_version = s->tablespace_version;
 
-	tablespace_version = fil_space_get_version(space);
-
-	low  = (offset / buf_read_ahead_random_area)
-		* buf_read_ahead_random_area;
-	high = (offset / buf_read_ahead_random_area + 1)
-		* buf_read_ahead_random_area;
-	if (high > fil_space_get_size(space)) {
+		low  = (offset / buf_read_ahead_random_area)
+			* buf_read_ahead_random_area;
+		high = (offset / buf_read_ahead_random_area + 1)
+			* buf_read_ahead_random_area;
+		high = s->max_page_number_for_io(high);
 
-		high = fil_space_get_size(space);
+		fil_space_release_for_io(s);
+	} else {
+		return 0;
 	}
 
 	buf_pool_mutex_enter(buf_pool);
@@ -435,22 +438,16 @@ buf_read_page(
 	ulint	zip_size,
 	ulint	offset)
 {
-	ib_int64_t	tablespace_version;
-	ulint		count;
 	dberr_t		err = DB_SUCCESS;
 
-	tablespace_version = fil_space_get_version(space_id);
-
 	FilSpace space(space_id, true);
 
 	if (space()) {
-
-		/* We do the i/o in the synchronous aio mode to save thread
-		switches: hence TRUE */
-		count = buf_read_page_low(&err, true, BUF_READ_ANY_PAGE, space_id,
-				  zip_size, FALSE,
-				  tablespace_version, offset);
-
+		ulint count = buf_read_page_low(&err, /*sync=*/true,
+						BUF_READ_ANY_PAGE,
+						space_id, zip_size, FALSE,
+						space()->tablespace_version,
+						offset);
 		srv_stats.buf_pool_reads.add(count);
 	}
 
@@ -619,21 +616,30 @@ buf_read_ahead_linear(
 		return(0);
 	}
 
-	/* Remember the tablespace version before we ask te tablespace size
-	below: if DISCARD + IMPORT changes the actual .ibd file meanwhile, we
-	do not try to read outside the bounds of the tablespace! */
+	uint32_t space_high_limit = 0;
 
-	tablespace_version = fil_space_get_version(space);
+	if (fil_space_t *s = fil_space_acquire_for_io(space)) {
+		/* Remember the tablespace version along with the
+		tablespace size: if DISCARD + IMPORT changes the
+		actual .ibd file meanwhile, we do not try to read
+		outside the bounds of the tablespace! */
+		tablespace_version = s->tablespace_version;
 
-	buf_pool_mutex_enter(buf_pool);
+		space_high_limit = s->max_page_number_for_io(ULINT_UNDEFINED);
 
-	if (high > fil_space_get_size(space)) {
-		buf_pool_mutex_exit(buf_pool);
+		fil_space_release_for_io(s);
+	} else {
+		return 0;
+	}
+
+	if (high > space_high_limit) {
 		/* The area is not whole, return */
 
 		return(0);
 	}
 
+	buf_pool_mutex_enter(buf_pool);
+
 	if (buf_pool->n_pend_reads
 	    > buf_pool->curr_size / BUF_READ_AHEAD_PEND_LIMIT) {
 		buf_pool_mutex_exit(buf_pool);
@@ -754,7 +760,7 @@ buf_read_ahead_linear(
 		return(0);
 	}
 
-	if (high > fil_space_get_size(space)) {
+	if (high > space_high_limit) {
 		/* The area is not whole, return */
 
 		return(0);

storage/innobase/fil/fil0fil.cc

@@ -676,7 +676,7 @@ fil_node_open_file(
 #ifdef UNIV_HOTBACKUP
 add_size:
 #endif /* UNIV_HOTBACKUP */
-		space->size += node->size;
+		space->committed_size = space->size += node->size;
 	}
 
 	ulint atomic_writes = FSP_FLAGS_GET_ATOMIC_WRITES(space->flags);
@@ -1151,6 +1151,9 @@ fil_mutex_enter_and_prepare_for_io(
 		ut_a(success);
 		/* InnoDB data files cannot shrink. */
 		ut_a(space->size >= size);
+		if (size > space->committed_size) {
+			space->committed_size = size;
+		}
 
 		/* There could be multiple concurrent I/O requests for
 		this tablespace (multiple threads trying to extend

storage/innobase/fsp/fsp0fsp.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -447,7 +447,7 @@ xdes_get_descriptor_with_space_hdr(
 	page_t*	descr_page;
 
 	ut_ad(mtr_memo_contains(mtr, fil_space_get_latch(space, NULL),
-				MTR_MEMO_X_LOCK));
+				MTR_MEMO_SPACE_X_LOCK));
 	ut_ad(mtr_memo_contains_page(mtr, sp_header, MTR_MEMO_PAGE_X_FIX));
 	ut_ad(page_offset(sp_header) == FSP_HEADER_OFFSET);
 	/* Read free limit and space size */
@@ -531,7 +531,7 @@ xdes_lst_get_descriptor(
 
 	ut_ad(mtr);
 	ut_ad(mtr_memo_contains(mtr, fil_space_get_latch(space, NULL),
-				MTR_MEMO_X_LOCK));
+				MTR_MEMO_SPACE_X_LOCK));
 	descr = fut_get_ptr(space, zip_size, lst_node, RW_X_LATCH, mtr)
 		- XDES_FLST_NODE;
 
@@ -687,7 +687,7 @@ fsp_header_init(ulint space_id, ulint size, mtr_t* mtr)
 
 	ut_ad(mtr);
 
-	mtr_x_lock(fil_space_get_latch(space_id, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space_id, &flags), mtr);
 
 	zip_size = fsp_flags_get_zip_size(flags);
 	block = buf_page_create(space_id, 0, zip_size, mtr);
@@ -814,7 +814,7 @@ fsp_header_inc_size(
 
 	ut_ad(mtr);
 
-	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, &flags), mtr);
 
 	header = fsp_get_space_header(space,
 				      fsp_flags_get_zip_size(flags),
@@ -843,7 +843,7 @@ fsp_header_get_tablespace_size(void)
 
 	mtr_start(&mtr);
 
-	mtr_x_lock(fil_space_get_latch(0, NULL), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(0, NULL), &mtr);
 
 	header = fsp_get_space_header(0, 0, &mtr);
 
@@ -1070,7 +1070,6 @@ fsp_fill_free_list(
 	ulint	frag_n_used;
 	ulint	actual_increase;
 	ulint	i;
-	mtr_t	ibuf_mtr;
 
 	ut_ad(page_offset(header) == FSP_HEADER_OFFSET);
 
@@ -1138,26 +1137,17 @@ fsp_fill_free_list(
 						 MLOG_2BYTES, mtr);
 			}
 
-			/* Initialize the ibuf bitmap page in a separate
-			mini-transaction because it is low in the latching
-			order, and we must be able to release its latch
-			before returning from the fsp routine */
-
-			mtr_start(&ibuf_mtr);
-
 			block = buf_page_create(space,
-						    i + FSP_IBUF_BITMAP_OFFSET,
-						    zip_size, &ibuf_mtr);
+						i + FSP_IBUF_BITMAP_OFFSET,
+						zip_size, mtr);
 			buf_page_get(space, zip_size,
 				     i + FSP_IBUF_BITMAP_OFFSET,
-				     RW_X_LATCH, &ibuf_mtr);
+				     RW_X_LATCH, mtr);
 			buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 
-			fsp_init_file_page(block, &ibuf_mtr);
+			fsp_init_file_page(block, mtr);
 
-			ibuf_bitmap_page_init(block, &ibuf_mtr);
-
-			mtr_commit(&ibuf_mtr);
+			ibuf_bitmap_page_init(block, mtr);
 		}
 
 		descr = xdes_get_descriptor_with_space_hdr(header, space, i,
@@ -2038,7 +2028,7 @@ fseg_create_general(
 		header = byte_offset + buf_block_get_frame(block);
 	}
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	if (!has_done_reservation) {
 		success = fsp_reserve_free_extents(&n_reserved, space, 2,
@@ -2185,7 +2175,7 @@ fseg_n_reserved_pages(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
 
@@ -2608,7 +2598,7 @@ fseg_alloc_free_page_general(
 
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	inode = fseg_inode_get(seg_header, space, zip_size, mtr);
 
@@ -2725,7 +2715,7 @@ fsp_reserve_free_extents(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	space_header = fsp_get_space_header(space, zip_size, mtr);
 try_again:
@@ -2865,7 +2855,7 @@ fsp_get_available_space_in_free_extents(
 
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	mutex_exit(&dict_sys->mutex);
 
@@ -3155,7 +3145,7 @@ fseg_free_page(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	seg_inode = fseg_inode_get(seg_header, space, zip_size, mtr);
 
@@ -3189,7 +3179,7 @@ fseg_page_is_free(
 	zip_size = dict_tf_get_zip_size(flags);
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	seg_inode = fseg_inode_get(seg_header, space, zip_size, &mtr);
 
@@ -3314,7 +3304,7 @@ fseg_free_step(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	descr = xdes_get_descriptor(space, zip_size, header_page, mtr);
 
@@ -3395,7 +3385,7 @@ fseg_free_step_not_header(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
 
@@ -3515,7 +3505,7 @@ fseg_validate_low(
 		ulint	zip_size;
 
 		mtr_start(&mtr);
-		mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+		mtr_x_space_lock(fil_space_get_latch(space, &flags), &mtr);
 		zip_size = fsp_flags_get_zip_size(flags);
 
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3538,7 +3528,7 @@ fseg_validate_low(
 		ulint	zip_size;
 
 		mtr_start(&mtr);
-		mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+		mtr_x_space_lock(fil_space_get_latch(space, &flags), &mtr);
 		zip_size = fsp_flags_get_zip_size(flags);
 
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3564,7 +3554,7 @@ fseg_validate_low(
 		ulint	zip_size;
 
 		mtr_start(&mtr);
-		mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+		mtr_x_space_lock(fil_space_get_latch(space, &flags), &mtr);
 		zip_size = fsp_flags_get_zip_size(flags);
 
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3602,7 +3592,7 @@ fseg_validate(
 
 	space = page_get_space_id(page_align(header));
 
-	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, &flags), mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
@@ -3678,7 +3668,7 @@ fseg_print(
 
 	space = page_get_space_id(page_align(header));
 
-	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, &flags), mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
@@ -3727,10 +3717,10 @@ fsp_validate(
 	/* Start first a mini-transaction mtr2 to lock out all other threads
 	from the fsp system */
 	mtr_start(&mtr2);
-	mtr_x_lock(latch, &mtr2);
+	mtr_x_space_lock(latch, &mtr2);
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -3757,7 +3747,7 @@ fsp_validate(
 
 	/* Validate FSP_FREE list */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 	node_addr = flst_get_first(header + FSP_FREE, &mtr);
@@ -3766,7 +3756,7 @@ fsp_validate(
 
 	while (!fil_addr_is_null(node_addr)) {
 		mtr_start(&mtr);
-		mtr_x_lock(latch, &mtr);
+		mtr_x_space_lock(latch, &mtr);
 
 		descr_count++;
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3781,7 +3771,7 @@ fsp_validate(
 
 	/* Validate FSP_FREE_FRAG list */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 	node_addr = flst_get_first(header + FSP_FREE_FRAG, &mtr);
@@ -3790,7 +3780,7 @@ fsp_validate(
 
 	while (!fil_addr_is_null(node_addr)) {
 		mtr_start(&mtr);
-		mtr_x_lock(latch, &mtr);
+		mtr_x_space_lock(latch, &mtr);
 
 		descr_count++;
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3808,7 +3798,7 @@ fsp_validate(
 
 	/* Validate FSP_FULL_FRAG list */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 	node_addr = flst_get_first(header + FSP_FULL_FRAG, &mtr);
@@ -3817,7 +3807,7 @@ fsp_validate(
 
 	while (!fil_addr_is_null(node_addr)) {
 		mtr_start(&mtr);
-		mtr_x_lock(latch, &mtr);
+		mtr_x_space_lock(latch, &mtr);
 
 		descr_count++;
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3832,7 +3822,7 @@ fsp_validate(
 
 	/* Validate segments */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -3847,7 +3837,7 @@ fsp_validate(
 		n = 0;
 		do {
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -3876,7 +3866,7 @@ fsp_validate(
 	}
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -3892,7 +3882,7 @@ fsp_validate(
 
 		do {
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -3976,11 +3966,11 @@ fsp_print(
 
 	mtr_start(&mtr2);
 
-	mtr_x_lock(latch, &mtr2);
+	mtr_x_space_lock(latch, &mtr2);
 
 	mtr_start(&mtr);
 
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -4012,7 +4002,7 @@ fsp_print(
 	/* Print segments */
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -4027,7 +4017,7 @@ fsp_print(
 		do {
 
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -4049,7 +4039,7 @@ fsp_print(
 	}
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -4064,7 +4054,7 @@ fsp_print(
 		do {
 
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -4121,7 +4111,8 @@ fsp_page_is_free_func(
 
 	ut_ad(mtr);
 
-	mtr_x_lock_func(fil_space_get_latch(space, &flags), file, line, mtr);
+	mtr_x_space_lock_func(fil_space_get_latch(space, &flags), file, line,
+			      mtr);
 	ulint zip_size = fsp_flags_get_zip_size(flags);
 
 	xdes_t* descr = xdes_get_descriptor(space, zip_size, page_no, mtr);

storage/innobase/ibuf/ibuf0ibuf.cc

@@ -548,7 +548,7 @@ ibuf_init_at_db_start(void)
 
 	mutex_enter(&ibuf_mutex);
 
-	mtr_x_lock(fil_space_get_latch(IBUF_SPACE_ID, NULL), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(IBUF_SPACE_ID, NULL), &mtr);
 
 	header_page = ibuf_header_page_get(&mtr);
 
@@ -2098,7 +2098,7 @@ ibuf_add_free_page(void)
 
 	/* Acquire the fsp latch before the ibuf header, obeying the latching
 	order */
-	mtr_x_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	header_page = ibuf_header_page_get(&mtr);
@@ -2182,7 +2182,7 @@ ibuf_remove_free_page(void)
 
 	/* Acquire the fsp latch before the ibuf header, obeying the latching
 	order */
-	mtr_x_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	header_page = ibuf_header_page_get(&mtr);

storage/innobase/include/fil0fil.h

@@ -306,6 +306,8 @@ struct fil_space_t {
 				/*!< recovered tablespace size in pages;
 				0 if no size change was read from the redo log,
 				or if the size change was implemented */
+  /** the committed size of the tablespace in pages */
+  ulint committed_size;
 	ulint		flags;	/*!< FSP_SPACE_FLAGS and FSP_FLAGS_MEM_ flags;
 				see fsp0fsp.h,
 				fsp_flags_is_valid(),
@@ -365,6 +367,15 @@ struct fil_space_t {
 	{
 		return stop_new_ops;
 	}
+
+  /** Clamp a page number for batched I/O, such as read-ahead.
+  @param offset   page number limit
+  @return offset clamped to the tablespace size */
+  ulint max_page_number_for_io(ulint offset) const
+  {
+    const ulint limit= committed_size;
+    return limit > offset ? offset : limit;
+  }
 };
 
 /** Value of fil_space_t::magic_n */

storage/innobase/include/mtr0mtr.h

@@ -2,7 +2,7 @@
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2013, 2016, MariaDB Corporation
+Copyright (c) 2013, 2020, 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
@@ -58,6 +58,7 @@ first 3 values must be RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH */
 #endif /* UNIV_DEBUG */
 #define	MTR_MEMO_S_LOCK		55
 #define	MTR_MEMO_X_LOCK		56
+#define	MTR_MEMO_SPACE_X_LOCK	57
 
 /** @name Log item types
 The log items are declared 'byte' so that the compiler can warn if val
@@ -295,6 +296,8 @@ This macro locks an rw-lock in s-mode. */
 This macro locks an rw-lock in x-mode. */
 #define mtr_x_lock(B, MTR)	mtr_x_lock_func((B), __FILE__, __LINE__,\
 						(MTR))
+#define mtr_x_space_lock(B, MTR) mtr_x_space_lock_func(B, __FILE__, __LINE__,\
+						       MTR)
 /*********************************************************************//**
 NOTE! Use the macro above!
 Locks a lock in s-mode. */
@@ -317,6 +320,14 @@ mtr_x_lock_func(
 	const char*	file,	/*!< in: file name */
 	ulint		line,	/*!< in: line number */
 	mtr_t*		mtr);	/*!< in: mtr */
+
+/** Acquire exclusive tablespace latch.
+@param lock   fil_space_t::latch
+@param file   source code file name of the caller
+@param line   source code line number of the caller
+@param mtr    mini-transaction */
+inline void mtr_x_space_lock_func(rw_lock_t *lock,
+                                  const char *file, ulint line, mtr_t *mtr);
 #endif /* !UNIV_HOTBACKUP */
 
 /***************************************************//**

storage/innobase/include/mtr0mtr.ic

 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2020, 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
@@ -78,7 +79,7 @@ mtr_memo_push(
 
 	ut_ad(object);
 	ut_ad(type >= MTR_MEMO_PAGE_S_FIX);
-	ut_ad(type <= MTR_MEMO_X_LOCK);
+	ut_ad(type <= MTR_MEMO_SPACE_X_LOCK);
 	ut_ad(mtr);
 	ut_ad(mtr->magic_n == MTR_MAGIC_N);
 	ut_ad(mtr->state == MTR_ACTIVE);
@@ -293,4 +294,16 @@ mtr_x_lock_func(
 
 	mtr_memo_push(mtr, lock, MTR_MEMO_X_LOCK);
 }
+
+/** Acquire exclusive tablespace latch.
+@param lock   fil_space_t::latch
+@param file   source code file name of the caller
+@param line   source code line number of the caller
+@param mtr    mini-transaction */
+inline void mtr_x_space_lock_func(rw_lock_t *lock,
+                                  const char *file, ulint line, mtr_t *mtr)
+{
+  rw_lock_x_lock_inline(lock, 0, file, line);
+  mtr_memo_push(mtr, lock, MTR_MEMO_SPACE_X_LOCK);
+}
 #endif /* !UNIV_HOTBACKUP */

storage/innobase/mtr/mtr0mtr.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2020, 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
@@ -87,6 +88,16 @@ mtr_memo_slot_release_func(
 	case MTR_MEMO_X_LOCK:
 		rw_lock_x_unlock((rw_lock_t*) object);
 		break;
+	case MTR_MEMO_SPACE_X_LOCK:
+		{
+			fil_space_t* space = reinterpret_cast<fil_space_t*>(
+				static_cast<char*>(object)
+				- reinterpret_cast<char*>(
+					&static_cast<fil_space_t*>(0)->latch));
+			space->committed_size = space->size;
+			rw_lock_x_unlock(&space->latch);
+		}
+		break;
 #ifdef UNIV_DEBUG
 	default:
 		ut_ad(slot->type == MTR_MEMO_MODIFY);

storage/innobase/srv/srv0start.cc

@@ -3,7 +3,7 @@
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All rights reserved.
 Copyright (c) 2008, Google Inc.
 Copyright (c) 2009, Percona Inc.
-Copyright (c) 2013, 2019, MariaDB Corporation.
+Copyright (c) 2013, 2020, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -1203,6 +1203,10 @@ open_or_create_data_files(
 		}
 	}
 
+	if (fil_space_t* s = fil_space_get(0)) {
+		s->committed_size = s->size;
+	}
+
 	return(DB_SUCCESS);
 }
 
@@ -1371,6 +1375,10 @@ srv_undo_tablespace_open(
 		if (fil_node_create(name, (ulint) n_pages, space, FALSE)) {
 			err = DB_SUCCESS;
 		}
+
+		if (fil_space_t* s = fil_space_get(space)) {
+			s->committed_size = n_pages;
+		}
 	}
 
 	return(err);

storage/innobase/trx/trx0rseg.cc

 /*****************************************************************************
 
 Copyright (c) 1996, 2011, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -65,7 +65,7 @@ trx_rseg_header_create(
 
 	ut_ad(mtr);
 	ut_ad(mtr_memo_contains(mtr, fil_space_get_latch(space, NULL),
-				MTR_MEMO_X_LOCK));
+				MTR_MEMO_SPACE_X_LOCK));
 
 	/* Allocate a new file segment for the rollback segment */
 	block = fseg_create(space, 0, TRX_RSEG + TRX_RSEG_FSEG_HEADER, mtr);
@@ -310,7 +310,7 @@ trx_rseg_create(ulint space)
 
 	/* To obey the latching order, acquire the file space
 	x-latch before the trx_sys->mutex. */
-	mtr_x_lock(fil_space_get_latch(space, NULL), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, NULL), &mtr);
 
 	slot_no = trx_sysf_rseg_find_free(&mtr);
 

storage/innobase/trx/trx0sys.cc

 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -550,7 +550,7 @@ trx_sysf_create(
 	then enter the kernel: we must do it in this order to conform
 	to the latching order rules. */
 
-	mtr_x_lock(fil_space_get_latch(TRX_SYS_SPACE, NULL), mtr);
+	mtr_x_space_lock(fil_space_get_latch(TRX_SYS_SPACE, NULL), mtr);
 
 	/* Create the trx sys file block in a new allocated file segment */
 	block = fseg_create(TRX_SYS_SPACE, 0, TRX_SYS + TRX_SYS_FSEG_HEADER,

storage/xtradb/buf/buf0flu.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2019, MariaDB Corporation.
+Copyright (c) 2013, 2020, MariaDB Corporation.
 Copyright (c) 2013, 2014, Fusion-io
 
 This program is free software; you can redistribute it and/or modify it under
@@ -1297,8 +1297,11 @@ buf_flush_try_neighbors(
 
 	/* fprintf(stderr, "Flush area: low %lu high %lu\n", low, high); */
 
-	if (high > fil_space_get_size(space)) {
-		high = fil_space_get_size(space);
+	if (fil_space_t *s = fil_space_acquire_for_io(space)) {
+		high = s->max_page_number_for_io(high);
+		fil_space_release_for_io(s);
+	} else {
+		return 0;
 	}
 
 	ulint	count = 0;

storage/xtradb/buf/buf0rea.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2013, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2018, MariaDB Corporation.
+Copyright (c) 2013, 2020, 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
@@ -344,19 +344,22 @@ buf_read_ahead_random(
 		return(0);
 	}
 
-	/* Remember the tablespace version before we ask te tablespace size
-	below: if DISCARD + IMPORT changes the actual .ibd file meanwhile, we
-	do not try to read outside the bounds of the tablespace! */
+	if (fil_space_t *s = fil_space_acquire_for_io(space)) {
+		/* Remember the tablespace version along with the
+		tablespace size: if DISCARD + IMPORT changes the
+		actual .ibd file meanwhile, we do not try to read
+		outside the bounds of the tablespace! */
+		tablespace_version = s->tablespace_version;
 
-	tablespace_version = fil_space_get_version(space);
-
-	low  = (offset / buf_read_ahead_random_area)
-		* buf_read_ahead_random_area;
-	high = (offset / buf_read_ahead_random_area + 1)
-		* buf_read_ahead_random_area;
-	if (high > fil_space_get_size(space)) {
+		low  = (offset / buf_read_ahead_random_area)
+			* buf_read_ahead_random_area;
+		high = (offset / buf_read_ahead_random_area + 1)
+			* buf_read_ahead_random_area;
+		high = s->max_page_number_for_io(high);
 
-		high = fil_space_get_size(space);
+		fil_space_release_for_io(s);
+	} else {
+		return 0;
 	}
 
 	if (buf_pool->n_pend_reads
@@ -495,22 +498,16 @@ buf_read_page(
 	ulint	offset,
 	trx_t*	trx)
 {
-	ib_int64_t	tablespace_version;
-	ulint		count;
 	dberr_t		err = DB_SUCCESS;
 
-	tablespace_version = fil_space_get_version(space_id);
-
 	FilSpace space(space_id, true);
 
 	if (space()) {
-
-		/* We do the i/o in the synchronous aio mode to save thread
-		switches: hence TRUE */
-		count = buf_read_page_low(&err, true, BUF_READ_ANY_PAGE, space_id,
-				  zip_size, FALSE,
-				  tablespace_version, offset, trx);
-
+		ulint count = buf_read_page_low(&err, /*sync=*/true,
+						BUF_READ_ANY_PAGE,
+						space_id, zip_size, FALSE,
+						space()->tablespace_version,
+						offset, trx);
 		srv_stats.buf_pool_reads.add(count);
 	}
 
@@ -683,13 +680,23 @@ buf_read_ahead_linear(
 		return(0);
 	}
 
-	/* Remember the tablespace version before we ask te tablespace size
-	below: if DISCARD + IMPORT changes the actual .ibd file meanwhile, we
-	do not try to read outside the bounds of the tablespace! */
+	uint32_t space_high_limit = 0;
 
-	tablespace_version = fil_space_get_version(space);
+	if (fil_space_t *s = fil_space_acquire_for_io(space)) {
+		/* Remember the tablespace version along with the
+		tablespace size: if DISCARD + IMPORT changes the
+		actual .ibd file meanwhile, we do not try to read
+		outside the bounds of the tablespace! */
+		tablespace_version = s->tablespace_version;
+
+		space_high_limit = s->max_page_number_for_io(ULINT_UNDEFINED);
+
+		fil_space_release_for_io(s);
+	} else {
+		return 0;
+	}
 
-	if (high > fil_space_get_size(space)) {
+	if (high > space_high_limit) {
 		/* The area is not whole, return */
 
 		return(0);
@@ -825,7 +832,7 @@ buf_read_ahead_linear(
 		return(0);
 	}
 
-	if (high > fil_space_get_size(space)) {
+	if (high > space_high_limit) {
 		/* The area is not whole, return */
 
 		return(0);

storage/xtradb/fil/fil0fil.cc

@@ -684,7 +684,7 @@ fil_node_open_file(
 #ifdef UNIV_HOTBACKUP
 add_size:
 #endif /* UNIV_HOTBACKUP */
-		space->size += node->size;
+		space->committed_size = space->size += node->size;
 	}
 
 	ulint atomic_writes = FSP_FLAGS_GET_ATOMIC_WRITES(space->flags);
@@ -1158,6 +1158,9 @@ fil_mutex_enter_and_prepare_for_io(
 		ut_a(success);
 		/* InnoDB data files cannot shrink. */
 		ut_a(space->size >= size);
+		if (size > space->committed_size) {
+			space->committed_size = size;
+		}
 
 		/* There could be multiple concurrent I/O requests for
 		this tablespace (multiple threads trying to extend

storage/xtradb/fsp/fsp0fsp.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -450,7 +450,7 @@ xdes_get_descriptor_with_space_hdr(
 	page_t*	descr_page;
 
 	ut_ad(mtr_memo_contains(mtr, fil_space_get_latch(space, NULL),
-				MTR_MEMO_X_LOCK));
+				MTR_MEMO_SPACE_X_LOCK));
 	ut_ad(mtr_memo_contains_page(mtr, sp_header, MTR_MEMO_PAGE_X_FIX));
 	ut_ad(page_offset(sp_header) == FSP_HEADER_OFFSET);
 	/* Read free limit and space size */
@@ -534,7 +534,7 @@ xdes_lst_get_descriptor(
 
 	ut_ad(mtr);
 	ut_ad(mtr_memo_contains(mtr, fil_space_get_latch(space, NULL),
-				MTR_MEMO_X_LOCK));
+				MTR_MEMO_SPACE_X_LOCK));
 	descr = fut_get_ptr(space, zip_size, lst_node, RW_X_LATCH, mtr)
 		- XDES_FLST_NODE;
 
@@ -690,7 +690,7 @@ fsp_header_init(ulint space_id, ulint size, mtr_t* mtr)
 
 	ut_ad(mtr);
 
-	mtr_x_lock(fil_space_get_latch(space_id, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space_id, &flags), mtr);
 
 	zip_size = fsp_flags_get_zip_size(flags);
 	block = buf_page_create(space_id, 0, zip_size, mtr);
@@ -817,7 +817,7 @@ fsp_header_inc_size(
 
 	ut_ad(mtr);
 
-	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, &flags), mtr);
 
 	header = fsp_get_space_header(space,
 				      fsp_flags_get_zip_size(flags),
@@ -846,7 +846,7 @@ fsp_header_get_tablespace_size(void)
 
 	mtr_start(&mtr);
 
-	mtr_x_lock(fil_space_get_latch(0, NULL), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(0, NULL), &mtr);
 
 	header = fsp_get_space_header(0, 0, &mtr);
 
@@ -1073,7 +1073,6 @@ fsp_fill_free_list(
 	ulint	frag_n_used;
 	ulint	actual_increase;
 	ulint	i;
-	mtr_t	ibuf_mtr;
 
 	ut_ad(page_offset(header) == FSP_HEADER_OFFSET);
 
@@ -1141,26 +1140,17 @@ fsp_fill_free_list(
 						 MLOG_2BYTES, mtr);
 			}
 
-			/* Initialize the ibuf bitmap page in a separate
-			mini-transaction because it is low in the latching
-			order, and we must be able to release its latch
-			before returning from the fsp routine */
-
-			mtr_start(&ibuf_mtr);
-
 			block = buf_page_create(space,
-						    i + FSP_IBUF_BITMAP_OFFSET,
-						    zip_size, &ibuf_mtr);
+						i + FSP_IBUF_BITMAP_OFFSET,
+						zip_size, mtr);
 			buf_page_get(space, zip_size,
 				     i + FSP_IBUF_BITMAP_OFFSET,
-				     RW_X_LATCH, &ibuf_mtr);
+				     RW_X_LATCH, mtr);
 			buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 
-			fsp_init_file_page(block, &ibuf_mtr);
+			fsp_init_file_page(block, mtr);
 
-			ibuf_bitmap_page_init(block, &ibuf_mtr);
-
-			mtr_commit(&ibuf_mtr);
+			ibuf_bitmap_page_init(block, mtr);
 		}
 
 		descr = xdes_get_descriptor_with_space_hdr(header, space, i,
@@ -2047,7 +2037,7 @@ fseg_create_general(
 		header = byte_offset + buf_block_get_frame(block);
 	}
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	if (!has_done_reservation) {
 		success = fsp_reserve_free_extents(&n_reserved, space, 2,
@@ -2194,7 +2184,7 @@ fseg_n_reserved_pages(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
 
@@ -2617,7 +2607,7 @@ fseg_alloc_free_page_general(
 
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	inode = fseg_inode_get(seg_header, space, zip_size, mtr);
 
@@ -2734,7 +2724,7 @@ fsp_reserve_free_extents(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	space_header = fsp_get_space_header(space, zip_size, mtr);
 try_again:
@@ -2873,7 +2863,7 @@ fsp_get_available_space_in_free_extents(
 
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	mutex_exit(&dict_sys->mutex);
 
@@ -3169,7 +3159,7 @@ fseg_free_page(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	seg_inode = fseg_inode_get(seg_header, space, zip_size, mtr);
 
@@ -3203,7 +3193,7 @@ fseg_page_is_free(
 	zip_size = dict_tf_get_zip_size(flags);
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	seg_inode = fseg_inode_get(seg_header, space, zip_size, &mtr);
 
@@ -3328,7 +3318,7 @@ fseg_free_step(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	descr = xdes_get_descriptor(space, zip_size, header_page, mtr);
 
@@ -3415,7 +3405,7 @@ fseg_free_step_not_header(
 	latch = fil_space_get_latch(space, &flags);
 	zip_size = fsp_flags_get_zip_size(flags);
 
-	mtr_x_lock(latch, mtr);
+	mtr_x_space_lock(latch, mtr);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
 
@@ -3541,7 +3531,7 @@ fseg_validate_low(
 		ulint	zip_size;
 
 		mtr_start(&mtr);
-		mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+		mtr_x_space_lock(fil_space_get_latch(space, &flags), &mtr);
 		zip_size = fsp_flags_get_zip_size(flags);
 
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3564,7 +3554,7 @@ fseg_validate_low(
 		ulint	zip_size;
 
 		mtr_start(&mtr);
-		mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+		mtr_x_space_lock(fil_space_get_latch(space, &flags), &mtr);
 		zip_size = fsp_flags_get_zip_size(flags);
 
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3590,7 +3580,7 @@ fseg_validate_low(
 		ulint	zip_size;
 
 		mtr_start(&mtr);
-		mtr_x_lock(fil_space_get_latch(space, &flags), &mtr);
+		mtr_x_space_lock(fil_space_get_latch(space, &flags), &mtr);
 		zip_size = fsp_flags_get_zip_size(flags);
 
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3628,7 +3618,7 @@ fseg_validate(
 
 	space = page_get_space_id(page_align(header));
 
-	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, &flags), mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
@@ -3704,7 +3694,7 @@ fseg_print(
 
 	space = page_get_space_id(page_align(header));
 
-	mtr_x_lock(fil_space_get_latch(space, &flags), mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, &flags), mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	inode = fseg_inode_get(header, space, zip_size, mtr);
@@ -3753,10 +3743,10 @@ fsp_validate(
 	/* Start first a mini-transaction mtr2 to lock out all other threads
 	from the fsp system */
 	mtr_start(&mtr2);
-	mtr_x_lock(latch, &mtr2);
+	mtr_x_space_lock(latch, &mtr2);
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -3783,7 +3773,7 @@ fsp_validate(
 
 	/* Validate FSP_FREE list */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 	node_addr = flst_get_first(header + FSP_FREE, &mtr);
@@ -3792,7 +3782,7 @@ fsp_validate(
 
 	while (!fil_addr_is_null(node_addr)) {
 		mtr_start(&mtr);
-		mtr_x_lock(latch, &mtr);
+		mtr_x_space_lock(latch, &mtr);
 
 		descr_count++;
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3807,7 +3797,7 @@ fsp_validate(
 
 	/* Validate FSP_FREE_FRAG list */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 	node_addr = flst_get_first(header + FSP_FREE_FRAG, &mtr);
@@ -3816,7 +3806,7 @@ fsp_validate(
 
 	while (!fil_addr_is_null(node_addr)) {
 		mtr_start(&mtr);
-		mtr_x_lock(latch, &mtr);
+		mtr_x_space_lock(latch, &mtr);
 
 		descr_count++;
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3834,7 +3824,7 @@ fsp_validate(
 
 	/* Validate FSP_FULL_FRAG list */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 	node_addr = flst_get_first(header + FSP_FULL_FRAG, &mtr);
@@ -3843,7 +3833,7 @@ fsp_validate(
 
 	while (!fil_addr_is_null(node_addr)) {
 		mtr_start(&mtr);
-		mtr_x_lock(latch, &mtr);
+		mtr_x_space_lock(latch, &mtr);
 
 		descr_count++;
 		descr = xdes_lst_get_descriptor(space, zip_size,
@@ -3858,7 +3848,7 @@ fsp_validate(
 
 	/* Validate segments */
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -3873,7 +3863,7 @@ fsp_validate(
 		n = 0;
 		do {
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -3902,7 +3892,7 @@ fsp_validate(
 	}
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -3918,7 +3908,7 @@ fsp_validate(
 
 		do {
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -4002,11 +3992,11 @@ fsp_print(
 
 	mtr_start(&mtr2);
 
-	mtr_x_lock(latch, &mtr2);
+	mtr_x_space_lock(latch, &mtr2);
 
 	mtr_start(&mtr);
 
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -4038,7 +4028,7 @@ fsp_print(
 	/* Print segments */
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -4053,7 +4043,7 @@ fsp_print(
 		do {
 
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -4075,7 +4065,7 @@ fsp_print(
 	}
 
 	mtr_start(&mtr);
-	mtr_x_lock(latch, &mtr);
+	mtr_x_space_lock(latch, &mtr);
 
 	header = fsp_get_space_header(space, zip_size, &mtr);
 
@@ -4090,7 +4080,7 @@ fsp_print(
 		do {
 
 			mtr_start(&mtr);
-			mtr_x_lock(latch, &mtr);
+			mtr_x_space_lock(latch, &mtr);
 
 			seg_inode_page = fut_get_ptr(
 				space, zip_size, node_addr, RW_X_LATCH, &mtr)
@@ -4147,7 +4137,8 @@ fsp_page_is_free_func(
 
 	ut_ad(mtr);
 
-	mtr_x_lock_func(fil_space_get_latch(space, &flags), file, line, mtr);
+	mtr_x_space_lock_func(fil_space_get_latch(space, &flags), file, line,
+			      mtr);
 	ulint zip_size = fsp_flags_get_zip_size(flags);
 
 	xdes_t* descr = xdes_get_descriptor(space, zip_size, page_no, mtr);

storage/xtradb/ibuf/ibuf0ibuf.cc

@@ -588,7 +588,7 @@ ibuf_init_at_db_start(void)
 
 	mutex_enter(&ibuf_mutex);
 
-	mtr_x_lock(fil_space_get_latch(IBUF_SPACE_ID, NULL), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(IBUF_SPACE_ID, NULL), &mtr);
 
 	header_page = ibuf_header_page_get(&mtr);
 
@@ -2139,7 +2139,7 @@ ibuf_add_free_page(void)
 
 	/* Acquire the fsp latch before the ibuf header, obeying the latching
 	order */
-	mtr_x_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	header_page = ibuf_header_page_get(&mtr);
@@ -2223,7 +2223,7 @@ ibuf_remove_free_page(void)
 
 	/* Acquire the fsp latch before the ibuf header, obeying the latching
 	order */
-	mtr_x_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(IBUF_SPACE_ID, &flags), &mtr);
 	zip_size = fsp_flags_get_zip_size(flags);
 
 	header_page = ibuf_header_page_get(&mtr);

storage/xtradb/include/fil0fil.h

@@ -300,6 +300,8 @@ struct fil_space_t {
 				/*!< recovered tablespace size in pages;
 				0 if no size change was read from the redo log,
 				or if the size change was implemented */
+  /** the committed size of the tablespace in pages */
+  ulint committed_size;
 	ulint		flags;	/*!< FSP_SPACE_FLAGS and FSP_FLAGS_MEM_ flags;
 				see fsp0fsp.h,
 				fsp_flags_is_valid(),
@@ -364,6 +366,15 @@ struct fil_space_t {
 	{
 		return stop_new_ops;
 	}
+
+  /** Clamp a page number for batched I/O, such as read-ahead.
+  @param offset   page number limit
+  @return offset clamped to the tablespace size */
+  ulint max_page_number_for_io(ulint offset) const
+  {
+    const ulint limit= committed_size;
+    return limit > offset ? offset : limit;
+  }
 };
 
 /** Value of fil_space_t::magic_n */

storage/xtradb/include/mtr0mtr.h

@@ -2,6 +2,7 @@
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
+Copyright (c) 2013, 2020, 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
@@ -57,6 +58,7 @@ first 3 values must be RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH */
 #endif /* UNIV_DEBUG */
 #define	MTR_MEMO_S_LOCK		55
 #define	MTR_MEMO_X_LOCK		56
+#define	MTR_MEMO_SPACE_X_LOCK	57
 
 /** @name Log item types
 The log items are declared 'byte' so that the compiler can warn if val
@@ -293,6 +295,8 @@ This macro locks an rw-lock in s-mode. */
 This macro locks an rw-lock in x-mode. */
 #define mtr_x_lock(B, MTR)	mtr_x_lock_func((B), __FILE__, __LINE__,\
 						(MTR))
+#define mtr_x_space_lock(B, MTR) mtr_x_space_lock_func(B, __FILE__, __LINE__,\
+						       MTR)
 /*********************************************************************//**
 NOTE! Use the macro above!
 Locks a lock in s-mode. */
@@ -315,6 +319,14 @@ mtr_x_lock_func(
 	const char*	file,	/*!< in: file name */
 	ulint		line,	/*!< in: line number */
 	mtr_t*		mtr);	/*!< in: mtr */
+
+/** Acquire exclusive tablespace latch.
+@param lock   fil_space_t::latch
+@param file   source code file name of the caller
+@param line   source code line number of the caller
+@param mtr    mini-transaction */
+inline void mtr_x_space_lock_func(prio_rw_lock_t *lock,
+                                  const char *file, ulint line, mtr_t *mtr);
 #endif /* !UNIV_HOTBACKUP */
 
 /***************************************************//**

storage/xtradb/include/mtr0mtr.ic

 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2020, 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
@@ -78,7 +79,7 @@ mtr_memo_push(
 
 	ut_ad(object);
 	ut_ad(type >= MTR_MEMO_PAGE_S_FIX);
-	ut_ad(type <= MTR_MEMO_X_LOCK);
+	ut_ad(type <= MTR_MEMO_SPACE_X_LOCK);
 	ut_ad(mtr);
 	ut_ad(mtr->magic_n == MTR_MAGIC_N);
 	ut_ad(mtr->state == MTR_ACTIVE);
@@ -293,4 +294,16 @@ mtr_x_lock_func(
 
 	mtr_memo_push(mtr, lock, MTR_MEMO_X_LOCK);
 }
+
+/** Acquire exclusive tablespace latch.
+@param lock   fil_space_t::latch
+@param file   source code file name of the caller
+@param line   source code line number of the caller
+@param mtr    mini-transaction */
+inline void mtr_x_space_lock_func(prio_rw_lock_t *lock,
+                                  const char *file, ulint line, mtr_t *mtr)
+{
+  rw_lock_x_lock_inline(lock, 0, file, line);
+  mtr_memo_push(mtr, lock, MTR_MEMO_SPACE_X_LOCK);
+}
 #endif /* !UNIV_HOTBACKUP */

storage/xtradb/mtr/mtr0mtr.cc

 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2020, 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
@@ -88,6 +89,16 @@ mtr_memo_slot_release_func(
 	case MTR_MEMO_X_LOCK:
 		rw_lock_x_unlock((prio_rw_lock_t*) object);
 		break;
+	case MTR_MEMO_SPACE_X_LOCK:
+		{
+			fil_space_t* space = reinterpret_cast<fil_space_t*>(
+				static_cast<char*>(object)
+				- reinterpret_cast<char*>(
+					&static_cast<fil_space_t*>(0)->latch));
+			space->committed_size = space->size;
+			rw_lock_x_unlock(&space->latch);
+		}
+		break;
 #ifdef UNIV_DEBUG
 	default:
 		ut_ad(slot->type == MTR_MEMO_MODIFY);

storage/xtradb/srv/srv0start.cc

@@ -3,7 +3,7 @@
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All rights reserved.
 Copyright (c) 2008, Google Inc.
 Copyright (c) 2009, Percona Inc.
-Copyright (c) 2013, 2019, MariaDB Corporation.
+Copyright (c) 2013, 2020, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -1240,6 +1240,10 @@ open_or_create_data_files(
 		}
 	}
 
+	if (fil_space_t* s = fil_space_get(0)) {
+		s->committed_size = s->size;
+	}
+
 	return(DB_SUCCESS);
 }
 
@@ -1408,6 +1412,10 @@ srv_undo_tablespace_open(
 		if (fil_node_create(name, (ulint) n_pages, space, FALSE)) {
 			err = DB_SUCCESS;
 		}
+
+		if (fil_space_t* s = fil_space_get(space)) {
+			s->committed_size = n_pages;
+		}
 	}
 
 	return(err);

storage/xtradb/trx/trx0rseg.cc

 /*****************************************************************************
 
 Copyright (c) 1996, 2011, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -65,7 +65,7 @@ trx_rseg_header_create(
 
 	ut_ad(mtr);
 	ut_ad(mtr_memo_contains(mtr, fil_space_get_latch(space, NULL),
-				MTR_MEMO_X_LOCK));
+				MTR_MEMO_SPACE_X_LOCK));
 
 	/* Allocate a new file segment for the rollback segment */
 	block = fseg_create(space, 0, TRX_RSEG + TRX_RSEG_FSEG_HEADER, mtr);
@@ -312,7 +312,7 @@ trx_rseg_create(ulint space)
 
 	/* To obey the latching order, acquire the file space
 	x-latch before the trx_sys->mutex. */
-	mtr_x_lock(fil_space_get_latch(space, NULL), &mtr);
+	mtr_x_space_lock(fil_space_get_latch(space, NULL), &mtr);
 
 	slot_no = trx_sysf_rseg_find_free(&mtr);
 

storage/xtradb/trx/trx0sys.cc

 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -554,7 +554,7 @@ trx_sysf_create(
 	then enter the kernel: we must do it in this order to conform
 	to the latching order rules. */
 
-	mtr_x_lock(fil_space_get_latch(TRX_SYS_SPACE, NULL), mtr);
+	mtr_x_space_lock(fil_space_get_latch(TRX_SYS_SPACE, NULL), mtr);
 
 	/* Create the trx sys file block in a new allocated file segment */
 	block = fseg_create(TRX_SYS_SPACE, 0, TRX_SYS + TRX_SYS_FSEG_HEADER,