MDEV-12467 encryption.create_or_replace hangs during DROP TABLE

fil_crypt_thread(): Do invoke fil_crypt_complete_rotate_space()
when the tablespace is about to be dropped. Also, remove a redundant
check whether rotate_thread_t::space is NULL. It can only become
NULL when fil_crypt_find_space_to_rotate() returns false, and in
that case we would already have terminated the loop.

fil_crypt_find_page_to_rotate(): Remove a redundant check for
space->crypt_data == NULL. Once encryption metadata has been
created for a tablespace, it cannot be removed without dropping
the entire tablespace.

storage/innobase/fil/fil0crypt.cc

@@ -1742,14 +1742,13 @@ fil_crypt_find_page_to_rotate(
 
 	fil_space_crypt_t *crypt_data = space->crypt_data;
 
-	/* Space might already be dropped */
-	if (crypt_data) {
 	mutex_enter(&crypt_data->mutex);
 	ut_ad(key_state->key_id == crypt_data->key_id);
 
-		if (crypt_data->rotate_state.next_offset <
-		    crypt_data->rotate_state.max_offset) {
+	bool found = crypt_data->rotate_state.max_offset >=
+		crypt_data->rotate_state.next_offset;
 
+	if (found) {
 		state->offset = crypt_data->rotate_state.next_offset;
 		ulint remaining = crypt_data->rotate_state.max_offset -
 			crypt_data->rotate_state.next_offset;
@@ -1759,16 +1758,11 @@ fil_crypt_find_page_to_rotate(
 		} else {
 			state->batch = remaining;
 		}
-
-			crypt_data->rotate_state.next_offset += batch;
-			mutex_exit(&crypt_data->mutex);
-			return true;
 	}
 
+	crypt_data->rotate_state.next_offset += batch;
 	mutex_exit(&crypt_data->mutex);
-	}
-
-	return false;
+	return found;
 }
 
 /***********************************************************************
@@ -2346,7 +2340,7 @@ DECLARE_THREAD(fil_crypt_thread)(
 			fil_crypt_start_rotate_space(&new_state, &thr);
 
 			/* iterate all pages (cooperativly with other threads) */
-			while (!thr.should_shutdown() && thr.space &&
+			while (!thr.should_shutdown() &&
 			       fil_crypt_find_page_to_rotate(&new_state, &thr)) {
 
 				/* rotate a (set) of pages */
@@ -2355,6 +2349,8 @@ DECLARE_THREAD(fil_crypt_thread)(
 				/* If space is marked as stopping, release
 				space and stop rotation. */
 				if (thr.space->is_stopping()) {
+					fil_crypt_complete_rotate_space(
+						&new_state, &thr);
 					fil_space_release(thr.space);
 					thr.space = NULL;
 					break;

storage/xtradb/fil/fil0crypt.cc

@@ -1742,14 +1742,13 @@ fil_crypt_find_page_to_rotate(
 
 	fil_space_crypt_t *crypt_data = space->crypt_data;
 
-	/* Space might already be dropped */
-	if (crypt_data) {
 	mutex_enter(&crypt_data->mutex);
 	ut_ad(key_state->key_id == crypt_data->key_id);
 
-		if (crypt_data->rotate_state.next_offset <
-		    crypt_data->rotate_state.max_offset) {
+	bool found = crypt_data->rotate_state.max_offset >=
+		crypt_data->rotate_state.next_offset;
 
+	if (found) {
 		state->offset = crypt_data->rotate_state.next_offset;
 		ulint remaining = crypt_data->rotate_state.max_offset -
 			crypt_data->rotate_state.next_offset;
@@ -1759,16 +1758,11 @@ fil_crypt_find_page_to_rotate(
 		} else {
 			state->batch = remaining;
 		}
-
-			crypt_data->rotate_state.next_offset += batch;
-			mutex_exit(&crypt_data->mutex);
-			return true;
 	}
 
+	crypt_data->rotate_state.next_offset += batch;
 	mutex_exit(&crypt_data->mutex);
-	}
-
-	return false;
+	return found;
 }
 
 /***********************************************************************
@@ -2346,7 +2340,7 @@ DECLARE_THREAD(fil_crypt_thread)(
 			fil_crypt_start_rotate_space(&new_state, &thr);
 
 			/* iterate all pages (cooperativly with other threads) */
-			while (!thr.should_shutdown() && thr.space &&
+			while (!thr.should_shutdown() &&
 			       fil_crypt_find_page_to_rotate(&new_state, &thr)) {
 
 				/* rotate a (set) of pages */
@@ -2355,6 +2349,8 @@ DECLARE_THREAD(fil_crypt_thread)(
 				/* If space is marked as stopping, release
 				space and stop rotation. */
 				if (thr.space->is_stopping()) {
+					fil_crypt_complete_rotate_space(
+						&new_state, &thr);
 					fil_space_release(thr.space);
 					thr.space = NULL;
 					break;