MDEV-14422 Assertion failure in trx_purge_run() on shutdown

row_quiesce_table_start(), row_quiesce_table_complete():
Use the more appropriate predicate srv_undo_sources for skipping
purge control. (This change alone is insufficient; it is possible
that this predicate will change during the call to trx_purge_stop()
or trx_purge_run().)

trx_purge_stop(), trx_purge_run(): Tolerate PURGE_STATE_EXIT.
It is very well possible to initiate shutdown soon after the statement
FLUSH TABLES FOR EXPORT has been submitted to execution.

srv_purge_coordinator_thread(): Ensure that the wait for purge_sys->event
in trx_purge_stop() will terminate when the coordinator thread exits.

storage/innobase/row/row0quiesce.cc

@@ -520,7 +520,7 @@ row_quiesce_table_start(
 	ut_ad(fil_space_get(table->space) != NULL);
 	ib::info() << "Sync to disk of " << table->name << " started.";
 
-	if (trx_purge_state() != PURGE_STATE_DISABLED) {
+	if (srv_undo_sources) {
 		trx_purge_stop();
 	}
 
@@ -603,7 +603,7 @@ row_quiesce_table_complete(
 		ib::info() << "Deleting the meta-data file '" << cfg_name << "'";
 	}
 
-	if (trx_purge_state() != PURGE_STATE_DISABLED) {
+	if (srv_undo_sources) {
 		trx_purge_run();
 	}
 

storage/innobase/srv/srv0srv.cc

@@ -2872,6 +2872,9 @@ DECLARE_THREAD(srv_purge_coordinator_thread)(
 
 	purge_sys->running = false;
 
+	/* Ensure that the wait in trx_purge_stop() will terminate. */
+	os_event_set(purge_sys->event);
+
 	rw_lock_x_unlock(&purge_sys->latch);
 
 #ifdef UNIV_DEBUG_THREAD_CREATION

storage/innobase/trx/trx0purge.cc

@@ -1770,52 +1770,48 @@ void
 trx_purge_stop(void)
 /*================*/
 {
-	ut_a(srv_n_purge_threads > 0);
-
 	rw_lock_x_lock(&purge_sys->latch);
 
-	const int64_t		sig_count = os_event_reset(purge_sys->event);
-	const purge_state_t	state = purge_sys->state;
-
-	ut_a(state == PURGE_STATE_RUN || state == PURGE_STATE_STOP);
-
-	++purge_sys->n_stop;
-
-	if (state == PURGE_STATE_RUN) {
+	switch (purge_sys->state) {
+	case PURGE_STATE_INIT:
+	case PURGE_STATE_DISABLED:
+		ut_error;
+	case PURGE_STATE_EXIT:
+		/* Shutdown must have been initiated during
+		FLUSH TABLES FOR EXPORT. */
+		ut_ad(!srv_undo_sources);
+unlock:
+		rw_lock_x_unlock(&purge_sys->latch);
+		break;
+	case PURGE_STATE_STOP:
+		ut_ad(srv_n_purge_threads > 0);
+		++purge_sys->n_stop;
+		purge_sys->state = PURGE_STATE_STOP;
+		if (!purge_sys->running) {
+			goto unlock;
+		}
+		ib::info() << "Waiting for purge to stop";
+		do {
+			rw_lock_x_unlock(&purge_sys->latch);
+			os_thread_sleep(10000);
+			rw_lock_x_lock(&purge_sys->latch);
+		} while (purge_sys->running);
+		goto unlock;
+	case PURGE_STATE_RUN:
+		ut_ad(srv_n_purge_threads > 0);
+		++purge_sys->n_stop;
 		ib::info() << "Stopping purge";
 
 		/* We need to wakeup the purge thread in case it is suspended,
 		so that it can acknowledge the state change. */
 
+		const int64_t sig_count = os_event_reset(purge_sys->event);
+		purge_sys->state = PURGE_STATE_STOP;
 		srv_purge_wakeup();
-	}
-
-	purge_sys->state = PURGE_STATE_STOP;
-
-	if (state != PURGE_STATE_STOP) {
 		rw_lock_x_unlock(&purge_sys->latch);
 		/* Wait for purge coordinator to signal that it
 		is suspended. */
 		os_event_wait_low(purge_sys->event, sig_count);
-	} else {
-		bool	once = true;
-
-		/* Wait for purge to signal that it has actually stopped. */
-		while (purge_sys->running) {
-
-			if (once) {
-				ib::info() << "Waiting for purge to stop";
-				once = false;
-			}
-
-			rw_lock_x_unlock(&purge_sys->latch);
-
-			os_thread_sleep(10000);
-
-			rw_lock_x_lock(&purge_sys->latch);
-		}
-
-		rw_lock_x_unlock(&purge_sys->latch);
 	}
 
 	MONITOR_INC_VALUE(MONITOR_PURGE_STOP_COUNT, 1);
@@ -1830,8 +1826,12 @@ trx_purge_run(void)
 	rw_lock_x_lock(&purge_sys->latch);
 
 	switch (purge_sys->state) {
-	case PURGE_STATE_INIT:
 	case PURGE_STATE_EXIT:
+		/* Shutdown must have been initiated during
+		FLUSH TABLES FOR EXPORT. */
+		ut_ad(!srv_undo_sources);
+		break;
+	case PURGE_STATE_INIT:
 	case PURGE_STATE_DISABLED:
 		ut_error;