InnoDB: implement innodb_max_purge_lag

innobase/include/srv0srv.h:
  Added srv_max_purge_lag and srv_dml_needed_delay
innobase/include/trx0sys.h:
  Added trx_sys->rseg_history_len
innobase/row/row0mysql.c:
  Added row_mysql_delay_if_needed()
innobase/srv/srv0srv.c:
  Added srv_max_purge_lag and srv_dml_needed_delay
innobase/trx/trx0purge.c:
  Update trx_sys->rseg_history_len.
  Calculate srv_dml_needed_delay from srv_max_purge_lag
  and trx_sys->rseg_history_len.
innobase/trx/trx0rseg.c:
  Initialize trx_sys->rseg_history_len
sql/ha_innodb.h:
  Add srv_max_purge_lag
sql/mysqld.cc:
  Add parameter innodb_max_purge_lag
sql/set_var.cc:
  Add global variable innodb_max_purge_lag

9 files changed, 82 insertions, 1 deletions

diff --git a/innobase/include/srv0srv.h b/innobase/include/srv0srv.h
index a5508b23a5c..2050e5b45c0 100644
--- a/innobase/include/srv0srv.h
+++ b/innobase/include/srv0srv.h
@@ -110,6 +110,8 @@ extern int      srv_query_thread_priority;
 
 extern ibool	srv_use_awe;
 extern ibool	srv_use_adaptive_hash_indexes;
+
+extern ulint	srv_max_purge_lag;
 /*-------------------------------------------*/
 
 extern ulint	srv_n_rows_inserted;
@@ -163,6 +165,7 @@ extern	ulint	srv_test_array_size;
 
 extern ulint	srv_activity_count;
 extern ulint	srv_fatal_semaphore_wait_threshold;
+extern ulint	srv_dml_needed_delay;
 
 extern mutex_t*	kernel_mutex_temp;/* mutex protecting the server, trx structs,
 				query threads, and lock table: we allocate
diff --git a/innobase/include/trx0sys.h b/innobase/include/trx0sys.h
index 8f402881224..31e8607f8a0 100644
--- a/innobase/include/trx0sys.h
+++ b/innobase/include/trx0sys.h
@@ -432,6 +432,10 @@ struct trx_sys_struct{
 	trx_rseg_t*	rseg_array[TRX_SYS_N_RSEGS];
 					/* Pointer array to rollback segments;
 					NULL if slot not in use */
+	ulint		rseg_history_len;/* Length of the TRX_RSEG_HISTORY
+					list (update undo logs for committed
+					transactions), protected by
+					rseg->mutex */
 	UT_LIST_BASE_NODE_T(read_view_t) view_list;
 					/* List of read views sorted on trx no,
 					biggest first */
diff --git a/innobase/row/row0mysql.c b/innobase/row/row0mysql.c
index 2e8c3adf94f..f4c4cc62082 100644
--- a/innobase/row/row0mysql.c
+++ b/innobase/row/row0mysql.c
@@ -92,6 +92,19 @@ row_mysql_is_system_table(
 	    || 0 == strcmp(name + 6, "user")
 	    || 0 == strcmp(name + 6, "db"));
 }
+
+/***********************************************************************
+Delays an INSERT, DELETE or UPDATE operation if the purge is lagging. */
+static
+void
+row_mysql_delay_if_needed(void)
+/*===========================*/
+{
+	if (srv_dml_needed_delay) {
+		os_thread_sleep(srv_dml_needed_delay);
+	}
+}
+
 /***********************************************************************
 Reads a MySQL format variable-length field (like VARCHAR) length and
 returns pointer to the field data. */
@@ -873,6 +886,8 @@ row_insert_for_mysql(
 
 	trx->op_info = "inserting";
 
+	row_mysql_delay_if_needed();
+
 	trx_start_if_not_started(trx);
 
 	if (node == NULL) {
@@ -1088,6 +1103,8 @@ row_update_for_mysql(
 
 	trx->op_info = "updating or deleting";
 
+	row_mysql_delay_if_needed();
+
 	trx_start_if_not_started(trx);
 
 	node = prebuilt->upd_node;
diff --git a/innobase/srv/srv0srv.c b/innobase/srv/srv0srv.c
index 9b96d610f28..adffe06ef0a 100644
--- a/innobase/srv/srv0srv.c
+++ b/innobase/srv/srv0srv.c
@@ -58,6 +58,10 @@ ulint	srv_activity_count	= 0;
 /* The following is the maximum allowed duration of a lock wait. */
 ulint	srv_fatal_semaphore_wait_threshold = 600;
 
+/* How much data manipulation language (DML) statements need to be delayed,
+in microseconds, in order to reduce the lagging of the purge thread. */
+ulint	srv_dml_needed_delay = 0;
+
 ibool	srv_lock_timeout_and_monitor_active = FALSE;
 ibool	srv_error_monitor_active = FALSE;
 
@@ -258,6 +262,8 @@ disable adaptive hash indexes */
 ibool	srv_use_awe			= FALSE;
 ibool	srv_use_adaptive_hash_indexes 	= TRUE;
 
+/* Maximum allowable purge history length.  <=0 means 'infinite'. */
+ulint	srv_max_purge_lag		= 0;
 
 /*-------------------------------------------*/
 ulint	srv_n_spin_wait_rounds	= 20;
diff --git a/innobase/trx/trx0purge.c b/innobase/trx/trx0purge.c
index 3d5f0d3f03a..1a4ef6a2e99 100644
--- a/innobase/trx/trx0purge.c
+++ b/innobase/trx/trx0purge.c
@@ -295,6 +295,9 @@ trx_purge_add_update_undo_to_history(
 	/* Add the log as the first in the history list */
 	flst_add_first(rseg_header + TRX_RSEG_HISTORY,
 				undo_header + TRX_UNDO_HISTORY_NODE, mtr);
+	mutex_enter(&kernel_mutex);
+	trx_sys->rseg_history_len++;
+	mutex_exit(&kernel_mutex);
 
 	/* Write the trx number to the undo log header */
 	mlog_write_dulint(undo_header + TRX_UNDO_TRX_NO, trx->no, mtr);
@@ -386,6 +389,12 @@ loop:
 
 	flst_cut_end(rseg_hdr + TRX_RSEG_HISTORY,
 			log_hdr + TRX_UNDO_HISTORY_NODE, n_removed_logs, &mtr);
+
+	mutex_enter(&kernel_mutex);
+	ut_ad(trx_sys->rseg_history_len >= n_removed_logs);
+	trx_sys->rseg_history_len -= n_removed_logs;
+	mutex_exit(&kernel_mutex);
+
 	freed = FALSE;
 
 	while (!freed) {
@@ -470,6 +479,11 @@ loop:
 	}
 
 	if (cmp >= 0) {
+		mutex_enter(&kernel_mutex);
+		ut_a(trx_sys->rseg_history_len >= n_removed_logs);
+		trx_sys->rseg_history_len -= n_removed_logs;
+		mutex_exit(&kernel_mutex);
+
 		flst_truncate_end(rseg_hdr + TRX_RSEG_HISTORY,
 	    			log_hdr + TRX_UNDO_HISTORY_NODE,
 				n_removed_logs, &mtr);
@@ -1031,6 +1045,27 @@ trx_purge(void)
 	purge_sys->view = NULL;
 	mem_heap_empty(purge_sys->heap);
 
+	/* Determine how much data manipulation language (DML) statements
+	need to be delayed in order to reduce the lagging of the purge
+	thread. */
+	srv_dml_needed_delay = 0; /* in microseconds; default: no delay */
+
+	/* If we cannot advance the 'purge view' because of an old
+	'consistent read view', then the DML statements cannot be delayed.
+	Also, srv_max_purge_lag <= 0 means 'infinity'. */
+	if (srv_max_purge_lag > 0
+			&& !UT_LIST_GET_LAST(trx_sys->view_list)) {
+		float	ratio = (float) trx_sys->rseg_history_len
+				/ srv_max_purge_lag;
+		if (ratio > 1) {
+			/* If the history list length exceeds the
+			innodb_max_purge_lag, the
+			data manipulation statements are delayed
+			by at least 5000 microseconds. */
+			srv_dml_needed_delay = (ratio - .5) * 10000;
+		}
+	}
+
 	purge_sys->view = read_view_oldest_copy_or_open_new(NULL,
 							purge_sys->heap);
 	mutex_exit(&kernel_mutex);	
diff --git a/innobase/trx/trx0rseg.c b/innobase/trx/trx0rseg.c
index e3885c86def..a01d4bb835d 100644
--- a/innobase/trx/trx0rseg.c
+++ b/innobase/trx/trx0rseg.c
@@ -135,6 +135,7 @@ trx_rseg_mem_create(
 	trx_ulogf_t*	undo_log_hdr;
 	fil_addr_t	node_addr;
 	ulint		sum_of_undo_sizes;
+	ulint		len;
 
 #ifdef UNIV_SYNC_DEBUG
 	ut_ad(mutex_own(&kernel_mutex));
@@ -166,7 +167,9 @@ trx_rseg_mem_create(
 							MLOG_4BYTES, mtr)
 			  + 1 + sum_of_undo_sizes;
 
-	if (flst_get_len(rseg_header + TRX_RSEG_HISTORY, mtr) > 0) {
+	len = flst_get_len(rseg_header + TRX_RSEG_HISTORY, mtr);
+	if (len > 0) {
+		trx_sys->rseg_history_len += len;
 
 		node_addr = trx_purge_get_log_from_hist(
 			      flst_get_last(rseg_header + TRX_RSEG_HISTORY,
@@ -206,6 +209,8 @@ trx_rseg_list_and_array_init(
 
 	UT_LIST_INIT(trx_sys->rseg_list);
 
+	trx_sys->rseg_history_len = 0;
+
 	for (i = 0; i < TRX_SYS_N_RSEGS; i++) {
 
 		page_no = trx_sysf_rseg_get_page_no(sys_header, i, mtr);
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index cab21b5bac4..6a4eb7571dc 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -194,6 +194,7 @@ extern my_bool innobase_log_archive,
 extern "C" {
 extern ulong srv_max_buf_pool_modified_pct;
 extern ulong srv_auto_extend_increment;
+extern ulong srv_max_purge_lag;
 }
 
 extern TYPELIB innobase_lock_typelib;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index f83dce39760..5e40398574b 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -3998,6 +3998,7 @@ enum options_mysqld
   OPT_INNODB_BUFFER_POOL_SIZE,
   OPT_INNODB_BUFFER_POOL_AWE_MEM_MB,
   OPT_INNODB_ADDITIONAL_MEM_POOL_SIZE,
+  OPT_INNODB_MAX_PURGE_LAG,
   OPT_INNODB_FILE_IO_THREADS,
   OPT_INNODB_LOCK_WAIT_TIMEOUT,
   OPT_INNODB_THREAD_CONCURRENCY,
@@ -4233,6 +4234,11 @@ Disable with --skip-innodb (will save memory).",
   {"innodb_max_dirty_pages_pct", OPT_INNODB_MAX_DIRTY_PAGES_PCT,
    "Percentage of dirty pages allowed in bufferpool.", (gptr*) &srv_max_buf_pool_modified_pct,
    (gptr*) &srv_max_buf_pool_modified_pct, 0, GET_ULONG, REQUIRED_ARG, 90, 0, 100, 0, 0, 0},
+  {"innodb_max_purge_lag", OPT_INNODB_MAX_PURGE_LAG,
+   "",
+   (gptr*) &srv_max_purge_lag,
+   (gptr*) &srv_max_purge_lag, 0, GET_LONG, REQUIRED_ARG, 0, 0, ~0L,
+   0, 1L, 0},
   {"innodb_status_file", OPT_INNODB_STATUS_FILE,
    "Enable SHOW INNODB STATUS output in the innodb_status.<pid> file",
    (gptr*) &innobase_create_status_file, (gptr*) &innobase_create_status_file,
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 3ea74a9f9e7..d661470576b 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -353,6 +353,8 @@ sys_var_long_ptr        sys_innodb_max_dirty_pages_pct("innodb_max_dirty_pages_p
                                                         &srv_max_buf_pool_modified_pct);
 sys_var_long_ptr	sys_innodb_autoextend_increment("innodb_autoextend_increment",
 							&srv_auto_extend_increment);
+sys_var_long_ptr	sys_innodb_max_purge_lag("innodb_max_purge_lag",
+							&srv_max_purge_lag);
 #endif
 
 /* Time/date/datetime formats */
@@ -603,6 +605,7 @@ sys_var *sys_variables[]=
   &sys_os,
 #ifdef HAVE_INNOBASE_DB
   &sys_innodb_max_dirty_pages_pct,
+  &sys_innodb_max_purge_lag,
   &sys_innodb_autoextend_increment,
 #endif    
   &sys_unique_checks,
@@ -697,6 +700,7 @@ struct show_var_st init_vars[]= {
   {"innodb_log_files_in_group", (char*) &innobase_log_files_in_group,	SHOW_LONG},
   {"innodb_log_group_home_dir", (char*) &innobase_log_group_home_dir, SHOW_CHAR_PTR},
   {sys_innodb_max_dirty_pages_pct.name, (char*) &sys_innodb_max_dirty_pages_pct, SHOW_SYS},
+  {sys_innodb_max_purge_lag.name, (char*) &sys_innodb_max_purge_lag, SHOW_SYS},
   {"innodb_mirrored_log_groups", (char*) &innobase_mirrored_log_groups, SHOW_LONG},
   {"innodb_open_files", (char*) &innobase_open_files, SHOW_LONG },
   {"innodb_thread_concurrency", (char*) &innobase_thread_concurrency, SHOW_LONG },