branches/5.1:

Fix Bug#38231 Innodb crash in lock_reset_all_on_table() on TRUNCATE + LOCK / UNLOCK

In TRUNCATE TABLE and discard tablespace: do not remove table-level S
and X locks and do not assert on such locks not being wait locks.
Leave such locks alone.

Approved by:	Heikki (rb://14)

include/lock0lock.h

@@ -463,14 +463,21 @@ void
 lock_cancel_waiting_and_release(
 /*============================*/
 	lock_t*	lock);	/* in: waiting lock request */
+
 /*************************************************************************
-Resets all locks, both table and record locks, on a table to be dropped.
-No lock is allowed to be a wait lock. */
+Removes locks on a table to be dropped or truncated.
+If remove_also_table_sx_locks is TRUE then table-level S and X locks are
+also removed in addition to other table-level and record-level locks.
+No lock, that is going to be removed, is allowed to be a wait lock. */
 
 void
-lock_reset_all_on_table(
-/*====================*/
-	dict_table_t*	table);	/* in: table to be dropped */
+lock_remove_all_on_table(
+/*=====================*/
+	dict_table_t*	table,			/* in: table to be dropped
+						or truncated */
+	ibool		remove_also_table_sx_locks);/* in: also removes
+						table S and X locks */
+
 /*************************************************************************
 Calculates the fold value of a page file address: used in inserting or
 searching for a lock in the hash table. */

lock/lock0lock.c

@@ -3920,15 +3920,25 @@ lock_cancel_waiting_and_release(
 	trx_end_lock_wait(lock->trx);
 }
 
+/* True if a lock mode is S or X */
+#define IS_LOCK_S_OR_X(lock) \
+	(lock_get_mode(lock) == LOCK_S \
+	 || lock_get_mode(lock) == LOCK_X)
+
+
 /*************************************************************************
-Resets all record and table locks of a transaction on a table to be dropped.
-No lock is allowed to be a wait lock. */
+Removes locks of a transaction on a table to be dropped.
+If remove_also_table_sx_locks is TRUE then table-level S and X locks are
+also removed in addition to other table-level and record-level locks.
+No lock, that is going to be removed, is allowed to be a wait lock. */
 static
 void
-lock_reset_all_on_table_for_trx(
-/*============================*/
-	dict_table_t*	table,	/* in: table to be dropped */
-	trx_t*		trx)	/* in: a transaction */
+lock_remove_all_on_table_for_trx(
+/*=============================*/
+	dict_table_t*	table,			/* in: table to be dropped */
+	trx_t*		trx,			/* in: a transaction */
+	ibool		remove_also_table_sx_locks)/* in: also removes
+						table S and X locks */
 {
 	lock_t*	lock;
 	lock_t*	prev_lock;
@@ -3946,7 +3956,9 @@ lock_reset_all_on_table_for_trx(
 
 			lock_rec_discard(lock);
 		} else if (lock_get_type(lock) & LOCK_TABLE
-			   && lock->un_member.tab_lock.table == table) {
+			   && lock->un_member.tab_lock.table == table
+			   && (remove_also_table_sx_locks
+			       || !IS_LOCK_S_OR_X(lock))) {
 
 			ut_a(!lock_get_wait(lock));
 
@@ -3958,26 +3970,65 @@ lock_reset_all_on_table_for_trx(
 }
 
 /*************************************************************************
-Resets all locks, both table and record locks, on a table to be dropped.
-No lock is allowed to be a wait lock. */
+Removes locks on a table to be dropped or truncated.
+If remove_also_table_sx_locks is TRUE then table-level S and X locks are
+also removed in addition to other table-level and record-level locks.
+No lock, that is going to be removed, is allowed to be a wait lock. */
 
 void
-lock_reset_all_on_table(
-/*====================*/
-	dict_table_t*	table)	/* in: table to be dropped */
+lock_remove_all_on_table(
+/*=====================*/
+	dict_table_t*	table,			/* in: table to be dropped
+						or truncated */
+	ibool		remove_also_table_sx_locks)/* in: also removes
+						table S and X locks */
 {
 	lock_t*	lock;
+	lock_t*	prev_lock;
 
 	mutex_enter(&kernel_mutex);
 
 	lock = UT_LIST_GET_FIRST(table->locks);
 
-	while (lock) {
-		ut_a(!lock_get_wait(lock));
+	while (lock != NULL) {
 
-		lock_reset_all_on_table_for_trx(table, lock->trx);
+		prev_lock = UT_LIST_GET_PREV(un_member.tab_lock.locks,
+					     lock);
 
-		lock = UT_LIST_GET_FIRST(table->locks);
+		/* If we should remove all locks (remove_also_table_sx_locks
+		is TRUE), or if the lock is not table-level S or X lock,
+		then check we are not going to remove a wait lock. */
+		if (remove_also_table_sx_locks
+		    || !(lock_get_type(lock) == LOCK_TABLE
+			 && IS_LOCK_S_OR_X(lock))) {
+
+			ut_a(!lock_get_wait(lock));
+		}
+
+		lock_remove_all_on_table_for_trx(table, lock->trx,
+						 remove_also_table_sx_locks);
+
+		if (prev_lock == NULL) {
+			if (lock == UT_LIST_GET_FIRST(table->locks)) {
+				/* lock was not removed, pick its successor */
+				lock = UT_LIST_GET_NEXT(
+					un_member.tab_lock.locks, lock);
+			} else {
+				/* lock was removed, pick the first one */
+				lock = UT_LIST_GET_FIRST(table->locks);
+			}
+		} else if (UT_LIST_GET_NEXT(un_member.tab_lock.locks,
+					    prev_lock) != lock) {
+			/* If lock was removed by
+			lock_remove_all_on_table_for_trx() then pick the
+			successor of prev_lock ... */
+			lock = UT_LIST_GET_NEXT(
+				un_member.tab_lock.locks, prev_lock);
+		} else {
+			/* ... otherwise pick the successor of lock. */
+			lock = UT_LIST_GET_NEXT(
+				un_member.tab_lock.locks, lock);
+		}
 	}
 
 	mutex_exit(&kernel_mutex);

mysql-test/innodb_bug38231.result

+SET storage_engine=InnoDB;
+INSERT INTO bug38231 VALUES (1), (10), (300);
+SET autocommit=0;
+SELECT * FROM bug38231 FOR UPDATE;
+a
+1
+10
+300
+TRUNCATE TABLE bug38231;
+COMMIT;
+DROP TABLE bug38231;

mysql-test/innodb_bug38231.test

+#
+# Bug#38231 Innodb crash in lock_reset_all_on_table() on TRUNCATE + LOCK / UNLOCK
+# http://bugs.mysql.com/38231
+#
+
+-- source include/have_innodb.inc
+
+SET storage_engine=InnoDB;
+
+# we care only that the following SQL commands do not crash the server
+-- disable_query_log
+-- disable_result_log
+
+DROP TABLE IF EXISTS bug38231;
+CREATE TABLE bug38231 (a INT);
+
+-- connect (con1,localhost,root,,)
+-- connect (con2,localhost,root,,)
+-- connect (con3,localhost,root,,)
+
+-- connection con1
+SET autocommit=0;
+LOCK TABLE bug38231 WRITE;
+
+-- connection con2
+SET autocommit=0;
+-- send
+LOCK TABLE bug38231 WRITE;
+
+-- connection con3
+SET autocommit=0;
+-- send
+LOCK TABLE bug38231 WRITE;
+
+-- connection default
+-- send
+TRUNCATE TABLE bug38231;
+
+-- connection con1
+# give time to TRUNCATE and others to be executed; without sleep, sometimes
+# UNLOCK executes before TRUNCATE
+-- sleep 0.2
+# this crashes the server if the bug is present
+UNLOCK TABLES;
+
+# clean up
+
+-- connection con2
+UNLOCK TABLES;
+
+-- connection con3
+UNLOCK TABLES;
+
+-- connection default
+
+-- disconnect con1
+-- disconnect con2
+-- disconnect con3
+
+# test that TRUNCATE works with with row-level locks
+
+-- enable_query_log
+-- enable_result_log
+
+INSERT INTO bug38231 VALUES (1), (10), (300);
+
+-- connect (con4,localhost,root,,)
+
+-- connection con4
+SET autocommit=0;
+SELECT * FROM bug38231 FOR UPDATE;
+
+-- connection default
+TRUNCATE TABLE bug38231;
+
+-- connection con4
+COMMIT;
+
+-- connection default
+
+-- disconnect con4
+
+DROP TABLE bug38231;

row/row0mysql.c

@@ -2451,8 +2451,8 @@ row_discard_tablespace_for_mysql(
 
 	new_id = dict_hdr_get_new_id(DICT_HDR_TABLE_ID);
 
-	/* Remove any locks there are on the table or its records */
-	lock_reset_all_on_table(table);
+	/* Remove all locks except the table-level S and X locks. */
+	lock_remove_all_on_table(table, FALSE);
 
 	info = pars_info_create();
 
@@ -2787,9 +2787,8 @@ row_truncate_table_for_mysql(
 		goto funct_exit;
 	}
 
-	/* Remove any locks there are on the table or its records */
-
-	lock_reset_all_on_table(table);
+	/* Remove all locks except the table-level S and X locks. */
+	lock_remove_all_on_table(table, FALSE);
 
 	trx->table_id = table->id;
 
@@ -3139,9 +3138,8 @@ check_next_foreign:
 		goto funct_exit;
 	}
 
-	/* Remove any locks there are on the table or its records */
-
-	lock_reset_all_on_table(table);
+	/* Remove all locks there are on the table or its records */
+	lock_remove_all_on_table(table, TRUE);
 
 	trx->dict_operation = TRUE;
 	trx->table_id = table->id;