MDEV-8182: Failing assertion: 1 == UT_LIST_GET_LEN(space->chain)

Analysis: At fil_spage_get_space there is small change that space
is found from tablespace list but we have not yet created node
for it (and added it to space->chain) and as we hold fil_system
mutex here fil_node_create can't continue.

Fixed by allowing UT_LIST_GET_LEN(space->chain) == 0|| 1 and
introducint two new functions that access filespace list
and before returning space check that node is also created.

storage/innobase/fil/fil0crypt.cc

@@ -1510,9 +1510,9 @@ fil_crypt_find_space_to_rotate(
 
 	if (state->first) {
 		state->first = false;
-		state->space = fil_get_first_space();
+		state->space = fil_get_first_space_safe();
 	} else {
-		state->space = fil_get_next_space(state->space);
+		state->space = fil_get_next_space_safe(state->space);
 	}
 
 	while (!state->should_shutdown() && state->space != ULINT_UNDEFINED) {
@@ -1525,7 +1525,7 @@ fil_crypt_find_space_to_rotate(
 			return true;
 		}
 
-		state->space = fil_get_next_space(state->space);
+		state->space = fil_get_next_space_safe(state->space);
 	}
 
 	/* if we didn't find any space return iops */
@@ -2047,7 +2047,6 @@ fil_crypt_complete_rotate_space(
 
 	/* Space might already be dropped */
 	if (crypt_data) {
-		ut_ad(crypt_data);
 		mutex_enter(&crypt_data->mutex);
 
 		/**
@@ -2096,6 +2095,7 @@ fil_crypt_complete_rotate_space(
 		if (btr_scrub_complete_space(&state->scrub_data) == true) {
 			if (should_flush) {
 				/* only last thread updates last_scrub_completed */
+				ut_ad(crypt_data);
 				mutex_enter(&crypt_data->mutex);
 				crypt_data->rotate_state.scrubbing.
 					last_scrub_completed = time(0);
@@ -2106,6 +2106,7 @@ fil_crypt_complete_rotate_space(
 		if (should_flush) {
 			fil_crypt_flush_space(state, space);
 
+			ut_ad(crypt_data);
 			mutex_enter(&crypt_data->mutex);
 			crypt_data->rotate_state.flushing = false;
 			mutex_exit(&crypt_data->mutex);

storage/innobase/fil/fil0fil.cc

@@ -1408,21 +1408,26 @@ fil_space_get_space(
 		}
 
 		/* The following code must change when InnoDB supports
-		multiple datafiles per tablespace. */
-		ut_a(1 == UT_LIST_GET_LEN(space->chain));
+		multiple datafiles per tablespace. Note that there is small
+		change that space is found from tablespace list but
+		we have not yet created node for it and as we hold
+		fil_system mutex here fil_node_create can't continue. */
+		ut_a(UT_LIST_GET_LEN(space->chain) == 1 || UT_LIST_GET_LEN(space->chain) == 0);
 
 		node = UT_LIST_GET_FIRST(space->chain);
 
-		/* It must be a single-table tablespace and we have not opened
-		the file yet; the following calls will open it and update the
-		size fields */
-
-		if (!fil_node_prepare_for_io(node, fil_system, space)) {
-			/* The single-table tablespace can't be opened,
-			because the ibd file is missing. */
-			return(NULL);
+		if (node) {
+			/* It must be a single-table tablespace and we have not opened
+			the file yet; the following calls will open it and update the
+			size fields */
+
+			if (!fil_node_prepare_for_io(node, fil_system, space)) {
+				/* The single-table tablespace can't be opened,
+				because the ibd file is missing. */
+				return(NULL);
+			}
+			fil_node_complete_io(node, fil_system, OS_FILE_READ);
 		}
-		fil_node_complete_io(node, fil_system, OS_FILE_READ);
 	}
 
 	return(space);
@@ -6764,6 +6769,7 @@ Get id of first tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
 fil_get_first_space()
+/*=================*/
 {
 	ulint out_id = ULINT_UNDEFINED;
 	fil_space_t* space;
@@ -6787,11 +6793,84 @@ fil_get_first_space()
 	return out_id;
 }
 
+/******************************************************************
+Get id of first tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_first_space_safe()
+/*======================*/
+{
+	ulint out_id = ULINT_UNDEFINED;
+	fil_space_t* space;
+
+	mutex_enter(&fil_system->mutex);
+
+	space = UT_LIST_GET_FIRST(fil_system->space_list);
+	if (space != NULL) {
+		do
+		{
+			if (!space->stop_new_ops && UT_LIST_GET_LEN(space->chain) > 0) {
+				out_id = space->id;
+				break;
+			}
+
+			space = UT_LIST_GET_NEXT(space_list, space);
+		} while (space != NULL);
+	}
+
+	mutex_exit(&fil_system->mutex);
+
+	return out_id;
+}
+
 /******************************************************************
 Get id of next tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
-fil_get_next_space(ulint id)
+fil_get_next_space(
+/*===============*/
+	ulint	id)	/*!< in: previous space id */
+{
+	bool found;
+	fil_space_t* space;
+	ulint out_id = ULINT_UNDEFINED;
+
+	mutex_enter(&fil_system->mutex);
+
+	space = fil_space_get_by_id(id);
+	if (space == NULL) {
+		/* we didn't find it...search for space with space->id > id */
+		found = false;
+		space = UT_LIST_GET_FIRST(fil_system->space_list);
+	} else {
+		/* we found it, take next available space */
+		found = true;
+	}
+
+	while ((space = UT_LIST_GET_NEXT(space_list, space)) != NULL) {
+
+		if (!found && space->id <= id)
+			continue;
+
+		if (!space->stop_new_ops && UT_LIST_GET_LEN(space->chain) > 0) {
+			/* inc reference to prevent drop */
+			out_id = space->id;
+			break;
+		}
+	}
+
+	mutex_exit(&fil_system->mutex);
+
+	return out_id;
+}
+
+/******************************************************************
+Get id of next tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_next_space_safe(
+/*====================*/
+	ulint	id)	/*!< in: previous space id */
 {
 	bool found;
 	fil_space_t* space;
@@ -6831,7 +6910,7 @@ Get crypt data for a tablespace */
 UNIV_INTERN
 fil_space_crypt_t*
 fil_space_get_crypt_data(
-/*==================*/
+/*=====================*/
 	ulint id)	/*!< in: space id */
 {
 	fil_space_t*	space;

storage/innobase/include/fil0fil.h

@@ -1264,14 +1264,32 @@ Get id of first tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
 fil_get_first_space();
+/*=================*/
 
 /******************************************************************
 Get id of next tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
 fil_get_next_space(
+/*===============*/
 	ulint id);      /*!< in: space id */
 
+/******************************************************************
+Get id of first tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_first_space_safe();
+/*======================*/
+
+/******************************************************************
+Get id of next tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_next_space_safe(
+/*====================*/
+	ulint	id);	/*!< in: previous space id */
+#endif /*  UNIV_INNOCHECKSUM */
+
 /*******************************************************************//**
 Returns the block size of the file space
 @return	block size */

storage/xtradb/fil/fil0crypt.cc

@@ -1510,9 +1510,9 @@ fil_crypt_find_space_to_rotate(
 
 	if (state->first) {
 		state->first = false;
-		state->space = fil_get_first_space();
+		state->space = fil_get_first_space_safe();
 	} else {
-		state->space = fil_get_next_space(state->space);
+		state->space = fil_get_next_space_safe(state->space);
 	}
 
 	while (!state->should_shutdown() && state->space != ULINT_UNDEFINED) {
@@ -1525,7 +1525,7 @@ fil_crypt_find_space_to_rotate(
 			return true;
 		}
 
-		state->space = fil_get_next_space(state->space);
+		state->space = fil_get_next_space_safe(state->space);
 	}
 
 	/* if we didn't find any space return iops */
@@ -2047,7 +2047,6 @@ fil_crypt_complete_rotate_space(
 
 	/* Space might already be dropped */
 	if (crypt_data) {
-		ut_ad(crypt_data);
 		mutex_enter(&crypt_data->mutex);
 
 		/**
@@ -2096,6 +2095,7 @@ fil_crypt_complete_rotate_space(
 		if (btr_scrub_complete_space(&state->scrub_data) == true) {
 			if (should_flush) {
 				/* only last thread updates last_scrub_completed */
+				ut_ad(crypt_data);
 				mutex_enter(&crypt_data->mutex);
 				crypt_data->rotate_state.scrubbing.
 					last_scrub_completed = time(0);
@@ -2106,6 +2106,7 @@ fil_crypt_complete_rotate_space(
 		if (should_flush) {
 			fil_crypt_flush_space(state, space);
 
+			ut_ad(crypt_data);
 			mutex_enter(&crypt_data->mutex);
 			crypt_data->rotate_state.flushing = false;
 			mutex_exit(&crypt_data->mutex);

storage/xtradb/fil/fil0fil.cc

@@ -1445,21 +1445,26 @@ fil_space_get_space(
 		}
 
 		/* The following code must change when InnoDB supports
-		multiple datafiles per tablespace. */
-		ut_a(1 == UT_LIST_GET_LEN(space->chain));
+		multiple datafiles per tablespace. Note that there is small
+		change that space is found from tablespace list but
+		we have not yet created node for it and as we hold
+		fil_system mutex here fil_node_create can't continue. */
+		ut_a(UT_LIST_GET_LEN(space->chain) == 1 || UT_LIST_GET_LEN(space->chain) == 0);
 
 		node = UT_LIST_GET_FIRST(space->chain);
 
-		/* It must be a single-table tablespace and we have not opened
-		the file yet; the following calls will open it and update the
-		size fields */
-
-		if (!fil_node_prepare_for_io(node, fil_system, space)) {
-			/* The single-table tablespace can't be opened,
-			because the ibd file is missing. */
-			return(NULL);
+		if (node) {
+			/* It must be a single-table tablespace and we have not opened
+			the file yet; the following calls will open it and update the
+			size fields */
+
+			if (!fil_node_prepare_for_io(node, fil_system, space)) {
+				/* The single-table tablespace can't be opened,
+				because the ibd file is missing. */
+				return(NULL);
+			}
+			fil_node_complete_io(node, fil_system, OS_FILE_READ);
 		}
-		fil_node_complete_io(node, fil_system, OS_FILE_READ);
 	}
 
 	return(space);
@@ -6891,6 +6896,7 @@ Get id of first tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
 fil_get_first_space()
+/*=================*/
 {
 	ulint out_id = ULINT_UNDEFINED;
 	fil_space_t* space;
@@ -6914,11 +6920,84 @@ fil_get_first_space()
 	return out_id;
 }
 
+/******************************************************************
+Get id of first tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_first_space_safe()
+/*======================*/
+{
+	ulint out_id = ULINT_UNDEFINED;
+	fil_space_t* space;
+
+	mutex_enter(&fil_system->mutex);
+
+	space = UT_LIST_GET_FIRST(fil_system->space_list);
+	if (space != NULL) {
+		do
+		{
+			if (!space->stop_new_ops && UT_LIST_GET_LEN(space->chain) > 0) {
+				out_id = space->id;
+				break;
+			}
+
+			space = UT_LIST_GET_NEXT(space_list, space);
+		} while (space != NULL);
+	}
+
+	mutex_exit(&fil_system->mutex);
+
+	return out_id;
+}
+
 /******************************************************************
 Get id of next tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
-fil_get_next_space(ulint id)
+fil_get_next_space(
+/*===============*/
+	ulint	id)	/*!< in: previous space id */
+{
+	bool found;
+	fil_space_t* space;
+	ulint out_id = ULINT_UNDEFINED;
+
+	mutex_enter(&fil_system->mutex);
+
+	space = fil_space_get_by_id(id);
+	if (space == NULL) {
+		/* we didn't find it...search for space with space->id > id */
+		found = false;
+		space = UT_LIST_GET_FIRST(fil_system->space_list);
+	} else {
+		/* we found it, take next available space */
+		found = true;
+	}
+
+	while ((space = UT_LIST_GET_NEXT(space_list, space)) != NULL) {
+
+		if (!found && space->id <= id)
+			continue;
+
+		if (!space->stop_new_ops && UT_LIST_GET_LEN(space->chain) > 0) {
+			/* inc reference to prevent drop */
+			out_id = space->id;
+			break;
+		}
+	}
+
+	mutex_exit(&fil_system->mutex);
+
+	return out_id;
+}
+
+/******************************************************************
+Get id of next tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_next_space_safe(
+/*====================*/
+	ulint	id)	/*!< in: previous space id */
 {
 	bool found;
 	fil_space_t* space;
@@ -6958,7 +7037,7 @@ Get crypt data for a tablespace */
 UNIV_INTERN
 fil_space_crypt_t*
 fil_space_get_crypt_data(
-/*==================*/
+/*=====================*/
 	ulint id)	/*!< in: space id */
 {
 	fil_space_t*	space;
@@ -6984,7 +7063,7 @@ Get crypt data for a tablespace */
 UNIV_INTERN
 fil_space_crypt_t*
 fil_space_set_crypt_data(
-/*==================*/
+/*=====================*/
 	ulint id, 	               /*!< in: space id */
 	fil_space_crypt_t* crypt_data) /*!< in: crypt data */
 {

storage/xtradb/include/fil0fil.h

@@ -1292,15 +1292,32 @@ Get id of first tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
 fil_get_first_space();
+/*=================*/
 
 /******************************************************************
 Get id of next tablespace or ULINT_UNDEFINED if none */
 UNIV_INTERN
 ulint
 fil_get_next_space(
+/*===============*/
 	ulint id);      /*!< in: space id */
 
-#endif
+/******************************************************************
+Get id of first tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_first_space_safe();
+/*======================*/
+
+/******************************************************************
+Get id of next tablespace that has node or ULINT_UNDEFINED if none */
+UNIV_INTERN
+ulint
+fil_get_next_space_safe(
+/*====================*/
+	ulint	id);	/*!< in: previous space id */
+
+#endif /*  UNIV_INNOCHECKSUM */
 
 /*******************************************************************//**
 Return space flags */