MDEV-31382 SET GLOBAL innodb_undo_log_truncate=ON has no effect on logically empty undo logs

innodb_undo_log_truncate_update(): A callback function. If
SET GLOBAL innodb_undo_log_truncate=ON, invoke
srv_wake_purge_thread_if_not_active().

srv_wake_purge_thread_if_not_active(): If innodb_undo_log_truncate=ON,
always wake up the purge subsystem.

srv_do_purge(): If the history is empty, invoke
trx_purge_truncate_history() in order to free undo log pages.

trx_purge_truncate_history(): If head.trx_no==0, consider the
cached undo logs to be free.

trx_purge(): Remove the parameter "bool truncate" and let the
caller invoke trx_purge_truncate_history() directly.

Reviewed by: Vladislav Lesin

storage/innobase/handler/ha_innodb.cc

@@ -19684,10 +19684,17 @@ static MYSQL_SYSVAR_ULONG(purge_rseg_truncate_frequency,
   " purge rollback segment(s) on every Nth iteration of purge invocation",
   NULL, NULL, 128, 1, 128, 0);
 
+static void innodb_undo_log_truncate_update(THD *thd, struct st_mysql_sys_var*,
+                                            void*, const void *save)
+{
+  if ((srv_undo_log_truncate= *static_cast<const my_bool*>(save)))
+    srv_wake_purge_thread_if_not_active();
+}
+
 static MYSQL_SYSVAR_BOOL(undo_log_truncate, srv_undo_log_truncate,
   PLUGIN_VAR_OPCMDARG,
   "Enable or Disable Truncate of UNDO tablespace.",
-  NULL, NULL, FALSE);
+  NULL, innodb_undo_log_truncate_update, FALSE);
 
 static MYSQL_SYSVAR_LONG(autoinc_lock_mode, innobase_autoinc_lock_mode,
   PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,

storage/innobase/include/trx0purge.h

@@ -43,12 +43,19 @@ Remove the undo log segment from the rseg slot if it is too big for reuse.
 @param[in,out]	mtr		mini-transaction */
 void
 trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr);
+
+/**
+Remove unnecessary history data from rollback segments. NOTE that when this
+function is called, the caller (purge_coordinator_callback)
+must not have any latches on undo log pages!
+*/
+void trx_purge_truncate_history();
+
 /**
 Run a purge batch.
 @param n_tasks   number of purge tasks to submit to the queue
-@param truncate  whether to truncate the history at the end of the batch
 @return number of undo log pages handled in the batch */
-ulint trx_purge(ulint n_tasks, bool truncate);
+ulint trx_purge(ulint n_tasks);
 
 /** Rollback segements from a given transaction with trx-no
 scheduled for purge. */

storage/innobase/srv/srv0srv.cc

@@ -1346,17 +1346,14 @@ static tpool::waitable_task purge_coordinator_task
 static tpool::timer *purge_coordinator_timer;
 
 /** Wake up the purge threads if there is work to do. */
-void
-srv_wake_purge_thread_if_not_active()
+void srv_wake_purge_thread_if_not_active()
 {
-	ut_ad(!srv_read_only_mode);
+  ut_ad(!srv_read_only_mode);
 
-	if (purge_sys.enabled() && !purge_sys.paused()
-	    && trx_sys.rseg_history_len) {
-		if(++purge_state.m_running == 1) {
-			srv_thread_pool->submit_task(&purge_coordinator_task);
-		}
-	}
+  if (purge_sys.enabled() && !purge_sys.paused() &&
+      (srv_undo_log_truncate || trx_sys.rseg_history_len) &&
+      ++purge_state.m_running == 1)
+    srv_thread_pool->submit_task(&purge_coordinator_task);
 }
 
 /** @return whether the purge tasks are active */
@@ -1811,8 +1808,8 @@ static size_t srv_do_purge(ulint* n_total_purged)
 			n_threads = n_use_threads = srv_n_purge_threads;
 			srv_purge_thread_count_changed = 0;
 		} else if (trx_sys.rseg_history_len > rseg_history_len
-		    || (srv_max_purge_lag > 0
-			&& rseg_history_len > srv_max_purge_lag)) {
+			   || (srv_max_purge_lag > 0
+			       && rseg_history_len > srv_max_purge_lag)) {
 
 			/* History length is now longer than what it was
 			when we took the last snapshot. Use more threads. */
@@ -1838,15 +1835,19 @@ static size_t srv_do_purge(ulint* n_total_purged)
 
 		/* Take a snapshot of the history list before purge. */
 		if (!(rseg_history_len = trx_sys.rseg_history_len)) {
-			break;
+			n_pages_purged = 0;
+			goto truncate;
 		}
 
-		n_pages_purged = trx_purge(
-			n_use_threads,
-			!(++count % srv_purge_rseg_truncate_frequency)
-			|| purge_sys.truncate.current
-			|| (srv_shutdown_state != SRV_SHUTDOWN_NONE
-			    && srv_fast_shutdown == 0));
+		n_pages_purged = trx_purge(n_use_threads);
+
+		if (!(++count % srv_purge_rseg_truncate_frequency)
+		    || purge_sys.truncate.current
+		    || (srv_shutdown_state != SRV_SHUTDOWN_NONE
+			&& srv_fast_shutdown == 0)) {
+truncate:
+			trx_purge_truncate_history();
+		}
 
 		*n_total_purged += n_pages_purged;
 	} while (n_pages_purged > 0 && !purge_sys.paused()

storage/innobase/trx/trx0purge.cc

@@ -522,10 +522,11 @@ __attribute__((optimize(0)))
 # endif
 #endif
 /**
-Removes unnecessary history data from rollback segments. NOTE that when this
-function is called, the caller must not have any latches on undo log pages!
+Remove unnecessary history data from rollback segments. NOTE that when this
+function is called, the caller (purge_coordinator_callback)
+must not have any latches on undo log pages!
 */
-static void trx_purge_truncate_history()
+void trx_purge_truncate_history()
 {
   ut_ad(purge_sys.head <= purge_sys.tail);
   purge_sys_t::iterator &head= purge_sys.head.trx_no
@@ -618,7 +619,7 @@ static void trx_purge_truncate_history()
       for (const trx_undo_t *undo= UT_LIST_GET_FIRST(rseg->undo_cached); undo;
            undo= UT_LIST_GET_NEXT(undo_list, undo))
       {
-        if (head.trx_no < undo->trx_id)
+        if (head.trx_no && head.trx_no < undo->trx_id)
           goto not_free;
         else
           cached+= undo->size;
@@ -731,7 +732,7 @@ static void trx_purge_truncate_history()
       ut_ad(rseg->id == i);
       ut_ad(rseg->is_persistent());
       ut_ad(!rseg->trx_ref_count);
-      ut_ad(rseg->needs_purge <= head.trx_no);
+      ut_ad(!head.trx_no || rseg->needs_purge <= head.trx_no);
       ut_d(const auto old_page= rseg->page_no);
 
       buf_block_t *rblock= trx_rseg_header_create(&space, i,
@@ -1235,9 +1236,8 @@ static void trx_purge_wait_for_workers_to_complete()
 /**
 Run a purge batch.
 @param n_tasks   number of purge tasks to submit to the queue
-@param truncate  whether to truncate the history at the end of the batch
 @return number of undo log pages handled in the batch */
-ulint trx_purge(ulint n_tasks, bool truncate)
+ulint trx_purge(ulint n_tasks)
 {
 	que_thr_t*	thr = NULL;
 	ulint		n_pages_handled;
@@ -1271,10 +1271,6 @@ ulint trx_purge(ulint n_tasks, bool truncate)
 
 	trx_purge_wait_for_workers_to_complete();
 
-	if (truncate) {
-		trx_purge_truncate_history();
-	}
-
 	MONITOR_INC_VALUE(MONITOR_PURGE_INVOKED, 1);
 	MONITOR_INC_VALUE(MONITOR_PURGE_N_PAGE_HANDLED, n_pages_handled);