1 files changed, 106 insertions, 162 deletions

diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 4b488c9f648..108e9182f00 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -41,8 +41,6 @@ Created 2/16/1996 Heikki Tuuri
 
 #include "my_global.h"
 
-#include "ha_prototypes.h"
-
 #include "mysqld.h"
 #include "mysql/psi/mysql_stage.h"
 #include "mysql/psi/psi.h"
@@ -76,7 +74,6 @@ Created 2/16/1996 Heikki Tuuri
 #include "srv0start.h"
 #include "srv0srv.h"
 #include "btr0defragment.h"
-#include "fsp0sysspace.h"
 #include "mysql/service_wsrep.h" /* wsrep_recovery */
 #include "trx0rseg.h"
 #include "os0proc.h"
@@ -104,7 +101,6 @@ Created 2/16/1996 Heikki Tuuri
 #include "zlib.h"
 #include "ut0crc32.h"
 #include "btr0scrub.h"
-#include "ut0new.h"
 
 /** Log sequence number immediately after startup */
 lsn_t	srv_start_lsn;
@@ -466,23 +462,16 @@ create_log_files(
 
 	const ulint size = ulint(srv_log_file_size >> srv_page_size_shift);
 
-	logfile0 = fil_node_create(
-		logfilename, size, log_space, false, false);
+	logfile0 = log_space->add(logfilename, OS_FILE_CLOSED, size,
+				  false, false)->name;
 	ut_a(logfile0);
 
 	for (unsigned i = 1; i < srv_n_log_files; i++) {
 
 		sprintf(logfilename + dirnamelen, "ib_logfile%u", i);
 
-		if (!fil_node_create(logfilename, size,
-				     log_space, false, false)) {
-
-			ib::error()
-				<< "Cannot create file node for log file "
-				<< logfilename;
-
-			return(DB_ERROR);
-		}
+		log_space->add(logfilename, OS_FILE_CLOSED, size,
+			       false, false);
 	}
 
 	log_sys.log.create(srv_n_log_files);
@@ -563,34 +552,6 @@ create_log_files_rename(
 }
 
 /*********************************************************************//**
-Opens a log file.
-@return DB_SUCCESS or error code */
-static MY_ATTRIBUTE((nonnull, warn_unused_result))
-dberr_t
-open_log_file(
-/*==========*/
-	pfs_os_file_t*	file,	/*!< out: file handle */
-	const char*	name,	/*!< in: log file name */
-	os_offset_t*	size)	/*!< out: file size */
-{
-	bool	ret;
-
-	*file = os_file_create(innodb_log_file_key, name,
-			       OS_FILE_OPEN, OS_FILE_AIO,
-			       OS_LOG_FILE, srv_read_only_mode, &ret);
-	if (!ret) {
-		ib::error() << "Unable to open '" << name << "'";
-		return(DB_ERROR);
-	}
-
-	*size = os_file_get_size(*file);
-
-	ret = os_file_close(*file);
-	ut_a(ret);
-	return(DB_SUCCESS);
-}
-
-/*********************************************************************//**
 Create undo tablespace.
 @return DB_SUCCESS or error code */
 static
@@ -658,83 +619,68 @@ srv_undo_tablespace_create(
 
 	return(err);
 }
-/*********************************************************************//**
-Open an undo tablespace.
-@return DB_SUCCESS or error code */
-static
-dberr_t
-srv_undo_tablespace_open(
-/*=====================*/
-	const char*	name,		/*!< in: tablespace file name */
-	ulint		space_id)	/*!< in: tablespace id */
+
+/** Open an undo tablespace.
+@param[in]	name		tablespace file name
+@param[in]	space_id	tablespace ID
+@param[in]	create_new_db	whether undo tablespaces are being created
+@return whether the tablespace was opened */
+static bool srv_undo_tablespace_open(const char* name, ulint space_id,
+				     bool create_new_db)
 {
 	pfs_os_file_t	fh;
-	bool		ret;
-	dberr_t		err	= DB_ERROR;
+	bool		success;
 	char		undo_name[sizeof "innodb_undo000"];
 
 	snprintf(undo_name, sizeof(undo_name),
 		 "innodb_undo%03u", static_cast<unsigned>(space_id));
 
-	if (!srv_file_check_mode(name)) {
-		ib::error() << "UNDO tablespaces must be " <<
-			(srv_read_only_mode ? "writable" : "readable") << "!";
-
-		return(DB_ERROR);
-	}
-
 	fh = os_file_create(
-		innodb_data_file_key, name,
-		OS_FILE_OPEN_RETRY
-		| OS_FILE_ON_ERROR_NO_EXIT
-		| OS_FILE_ON_ERROR_SILENT,
-		OS_FILE_NORMAL,
-		OS_DATA_FILE,
-		srv_read_only_mode,
-		&ret);
-
-	/* If the file open was successful then load the tablespace. */
-
-	if (ret) {
-		os_offset_t	size;
-		fil_space_t*	space;
-
-		size = os_file_get_size(fh);
-		ut_a(size != (os_offset_t) -1);
+		innodb_data_file_key, name, OS_FILE_OPEN
+		| OS_FILE_ON_ERROR_NO_EXIT | OS_FILE_ON_ERROR_SILENT,
+		OS_FILE_AIO, OS_DATA_FILE, srv_read_only_mode, &success);
+	if (!success) {
+		return false;
+	}
 
-		ret = os_file_close(fh);
-		ut_a(ret);
+	os_offset_t size = os_file_get_size(fh);
+	ut_a(size != os_offset_t(-1));
 
-		/* Load the tablespace into InnoDB's internal
-		data structures. */
+	/* Load the tablespace into InnoDB's internal data structures. */
 
-		/* We set the biggest space id to the undo tablespace
-		because InnoDB hasn't opened any other tablespace apart
-		from the system tablespace. */
+	/* We set the biggest space id to the undo tablespace
+	because InnoDB hasn't opened any other tablespace apart
+	from the system tablespace. */
 
-		fil_set_max_space_id_if_bigger(space_id);
+	fil_set_max_space_id_if_bigger(space_id);
 
-		space = fil_space_create(
-			undo_name, space_id, FSP_FLAGS_PAGE_SSIZE(),
-			FIL_TYPE_TABLESPACE, NULL);
+	fil_space_t* space = fil_space_create(
+		undo_name, space_id, FSP_FLAGS_PAGE_SSIZE(),
+		FIL_TYPE_TABLESPACE, NULL);
 
-		ut_a(fil_validate());
-		ut_a(space);
+	ut_a(fil_validate());
+	ut_a(space);
 
-		os_offset_t	n_pages = size >> srv_page_size_shift;
+	fil_node_t* file = space->add(name, fh, 0, false, true);
 
-		/* On 32-bit platforms, ulint is 32 bits and os_offset_t
-		is 64 bits. It is OK to cast the n_pages to ulint because
-		the unit has been scaled to pages and page number is always
-		32 bits. */
-		if (fil_node_create(
-			name, (ulint) n_pages, space, false, TRUE)) {
+	mutex_enter(&fil_system.mutex);
 
-			err = DB_SUCCESS;
+	if (create_new_db) {
+		space->size = file->size = ulint(size >> srv_page_size_shift);
+		space->size_in_header = SRV_UNDO_TABLESPACE_SIZE_IN_PAGES;
+	} else {
+		success = file->read_page0(true);
+		if (!success) {
+			os_file_close(file->handle);
+			file->handle = OS_FILE_CLOSED;
+			ut_a(fil_system.n_open > 0);
+			fil_system.n_open--;
 		}
 	}
 
-	return(err);
+	mutex_exit(&fil_system.mutex);
+
+	return success;
 }
 
 /** Check if undo tablespaces and redo log files exist before creating a
@@ -920,12 +866,11 @@ srv_undo_tablespaces_init(bool create_new_db)
 		ut_a(undo_tablespace_ids[i] != 0);
 		ut_a(undo_tablespace_ids[i] != ULINT_UNDEFINED);
 
-		err = srv_undo_tablespace_open(name, undo_tablespace_ids[i]);
-
-		if (err != DB_SUCCESS) {
+		if (!srv_undo_tablespace_open(name, undo_tablespace_ids[i],
+					      create_new_db)) {
 			ib::error() << "Unable to open undo tablespace '"
 				<< name << "'.";
-			return(err);
+			return DB_ERROR;
 		}
 
 		prev_space_id = undo_tablespace_ids[i];
@@ -950,9 +895,8 @@ srv_undo_tablespaces_init(bool create_new_db)
 			name, sizeof(name),
 			"%s%cundo%03zu", srv_undo_dir, OS_PATH_SEPARATOR, i);
 
-		err = srv_undo_tablespace_open(name, i);
-
-		if (err != DB_SUCCESS) {
+		if (!srv_undo_tablespace_open(name, i, create_new_db)) {
+			err = DB_ERROR;
 			break;
 		}
 
@@ -1656,8 +1600,9 @@ dberr_t srv_start(bool create_new_db)
 			return(srv_init_abort(err));
 		}
 	} else {
+		srv_log_file_size = 0;
+
 		for (i = 0; i < SRV_N_LOG_FILES_MAX; i++) {
-			os_offset_t	size;
 			os_file_stat_t	stat_info;
 
 			sprintf(logfilename + dirnamelen,
@@ -1675,56 +1620,21 @@ dberr_t srv_start(bool create_new_db)
 					    == SRV_OPERATION_RESTORE_EXPORT) {
 						return(DB_SUCCESS);
 					}
-					if (flushed_lsn
-					    < static_cast<lsn_t>(1000)) {
-						ib::error()
-							<< "Cannot create"
-							" log files because"
-							" data files are"
-							" corrupt or the"
-							" database was not"
-							" shut down cleanly"
-							" after creating"
-							" the data files.";
-						return(srv_init_abort(
-							DB_ERROR));
-					}
-
-					err = create_log_files(
-						logfilename, dirnamelen,
-						flushed_lsn, logfile0);
-
-					if (err == DB_SUCCESS) {
-						err = create_log_files_rename(
-							logfilename,
-							dirnamelen,
-							flushed_lsn, logfile0);
-					}
-
-					if (err != DB_SUCCESS) {
-						return(srv_init_abort(err));
-					}
-
-					/* Suppress the message about
-					crash recovery. */
-					flushed_lsn = log_get_lsn();
-					goto files_checked;
 				}
 
 				/* opened all files */
 				break;
 			}
 
-			if (!srv_file_check_mode(logfilename)) {
-				return(srv_init_abort(DB_ERROR));
+			if (stat_info.type != OS_FILE_TYPE_FILE) {
+				break;
 			}
 
-			err = open_log_file(&files[i], logfilename, &size);
-
-			if (err != DB_SUCCESS) {
-				return(srv_init_abort(err));
+			if (!srv_file_check_mode(logfilename)) {
+				return(srv_init_abort(DB_ERROR));
 			}
 
+			const os_offset_t size = stat_info.size;
 			ut_a(size != (os_offset_t) -1);
 
 			if (size & (OS_FILE_LOG_BLOCK_SIZE - 1)) {
@@ -1749,8 +1659,13 @@ dberr_t srv_start(bool create_new_db)
 				/* The first log file must consist of
 				at least the following 512-byte pages:
 				header, checkpoint page 1, empty,
-				checkpoint page 2, redo log page(s) */
-				if (size <= OS_FILE_LOG_BLOCK_SIZE * 4) {
+				checkpoint page 2, redo log page(s).
+
+				Mariabackup --prepare would create an
+				empty ib_logfile0. Tolerate it if there
+				are no other ib_logfile* files. */
+				if ((size != 0 || i != 0)
+				    && size <= OS_FILE_LOG_BLOCK_SIZE * 4) {
 					ib::error() << "Log file "
 						<< logfilename << " size "
 						<< size << " is too small";
@@ -1767,6 +1682,39 @@ dberr_t srv_start(bool create_new_db)
 			}
 		}
 
+		if (srv_log_file_size == 0) {
+			if (flushed_lsn < lsn_t(1000)) {
+				ib::error()
+					<< "Cannot create log files because"
+					" data files are corrupt or the"
+					" database was not shut down cleanly"
+					" after creating the data files.";
+				return srv_init_abort(DB_ERROR);
+			}
+
+			strcpy(logfilename + dirnamelen, "ib_logfile0");
+			srv_log_file_size = srv_log_file_size_requested;
+
+			err = create_log_files(
+				logfilename, dirnamelen,
+				flushed_lsn, logfile0);
+
+			if (err == DB_SUCCESS) {
+				err = create_log_files_rename(
+					logfilename, dirnamelen,
+					flushed_lsn, logfile0);
+			}
+
+			if (err != DB_SUCCESS) {
+				return(srv_init_abort(err));
+			}
+
+			/* Suppress the message about
+			crash recovery. */
+			flushed_lsn = log_get_lsn();
+			goto files_checked;
+		}
+
 		srv_n_log_files_found = i;
 
 		/* Create the in-memory file space objects. */
@@ -1791,10 +1739,8 @@ dberr_t srv_start(bool create_new_db)
 		for (unsigned j = 0; j < srv_n_log_files_found; j++) {
 			sprintf(logfilename + dirnamelen, "ib_logfile%u", j);
 
-			if (!fil_node_create(logfilename, size,
-					     log_space, false, false)) {
-				return(srv_init_abort(DB_ERROR));
-			}
+			log_space->add(logfilename, OS_FILE_CLOSED, size,
+				       false, false);
 		}
 
 		log_sys.log.create(srv_n_log_files_found);
@@ -2130,24 +2076,24 @@ files_checked:
 				page_id_t(IBUF_SPACE_ID,
 					  FSP_IBUF_HEADER_PAGE_NO),
 				univ_page_size, RW_X_LATCH, &mtr);
-			fil_block_check_type(block, FIL_PAGE_TYPE_SYS, &mtr);
+			fil_block_check_type(*block, FIL_PAGE_TYPE_SYS, &mtr);
 			/* Already MySQL 3.23.53 initialized
 			FSP_IBUF_TREE_ROOT_PAGE_NO to
 			FIL_PAGE_INDEX. No need to reset that one. */
 			block = buf_page_get(
 				page_id_t(TRX_SYS_SPACE, TRX_SYS_PAGE_NO),
 				univ_page_size, RW_X_LATCH, &mtr);
-			fil_block_check_type(block, FIL_PAGE_TYPE_TRX_SYS,
+			fil_block_check_type(*block, FIL_PAGE_TYPE_TRX_SYS,
 					     &mtr);
 			block = buf_page_get(
 				page_id_t(TRX_SYS_SPACE,
 					  FSP_FIRST_RSEG_PAGE_NO),
 				univ_page_size, RW_X_LATCH, &mtr);
-			fil_block_check_type(block, FIL_PAGE_TYPE_SYS, &mtr);
+			fil_block_check_type(*block, FIL_PAGE_TYPE_SYS, &mtr);
 			block = buf_page_get(
 				page_id_t(TRX_SYS_SPACE, FSP_DICT_HDR_PAGE_NO),
 				univ_page_size, RW_X_LATCH, &mtr);
-			fil_block_check_type(block, FIL_PAGE_TYPE_SYS, &mtr);
+			fil_block_check_type(*block, FIL_PAGE_TYPE_SYS, &mtr);
 			mtr.commit();
 		}
 
@@ -2386,7 +2332,7 @@ files_checked:
 		  Create the dump/load thread only when not running with
 		  --wsrep-recover.
 		*/
-		if (!wsrep_recovery) {
+		if (!get_wsrep_recovery()) {
 #endif /* WITH_WSREP */
 
 		/* Create the buffer pool dump/load thread */
@@ -2474,9 +2420,7 @@ void srv_shutdown_bg_undo_sources()
 /** Shut down InnoDB. */
 void innodb_shutdown()
 {
-	ut_ad(!my_atomic_loadptr_explicit(reinterpret_cast<void**>
-					  (&srv_running),
-					  MY_MEMORY_ORDER_RELAXED));
+	ut_ad(!srv_running.load(std::memory_order_relaxed));
 	ut_ad(!srv_undo_sources);
 
 	switch (srv_operation) {