Merge 10.2

mysql-test/r/xa.result

@@ -239,9 +239,7 @@ XA END 'xa1';
 XA ROLLBACK 'xa1';
 DROP TABLE t1;
 #
-# Bug#12352846 - TRANS_XA_START(THD*):
-#                ASSERTION THD->TRANSACTION.XID_STATE.XID.IS_NULL()
-#                FAILED 
+# MDEV-10962 Deadlock with 3 concurrent DELETEs by unique key
 #
 DROP TABLE IF EXISTS t1, t2;
 CREATE TABLE t1 (a INT) ENGINE=InnoDB;
@@ -255,13 +253,12 @@ INSERT INTO t2 SELECT a FROM t1;
 connection default;
 # Waiting until INSERT ... is blocked
 DELETE FROM t1;
+COMMIT;
 connection con2;
 # Reaping: INSERT INTO t2 SELECT a FROM t1
-ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+XA END 'xid1';
+XA PREPARE 'xid1';
 XA COMMIT 'xid1';
-ERROR XA102: XA_RBDEADLOCK: Transaction branch was rolled back: deadlock was detected
-connection default;
-COMMIT;
 connection con2;
 XA START 'xid1';
 XA END 'xid1';

mysql-test/suite/innodb/r/innodb-lock.result

@@ -159,3 +159,174 @@ connection con1;
 disconnect con1;
 connection default;
 DROP TABLE t1, t2, t3;
+#
+# MDEV-18706 ER_LOCK_DEADLOCK on concurrent read and insert into already locked gap
+#
+# A) Def inserts between infimum and Def-locked record, between Def-locked record and supremum
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (3, 0);
+start transaction;
+update t1 set x= 1;
+connect  con1, localhost, root,, test;
+select * from t1 for update;
+connection default;
+insert into t1 values (1, 1);
+insert into t1 values (2, 1);
+insert into t1 values (4, 1);
+insert into t1 values (5, 1);
+commit;
+connection con1;
+# No ER_LOCK_DEADLOCK; all rows returned
+pk	x
+1	1
+2	1
+3	1
+4	1
+5	1
+connection default;
+disconnect con1;
+drop table t1;
+# B) Def inserts between Con1-locked and Def-locked record
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (1, 2);
+insert into t1 values (4, 0);
+start transaction;
+update t1 set x= 2 where pk > 3;
+connect  con1,localhost,root,,test;
+select * from t1 for update;
+connection default;
+insert into t1 values (2, 2);
+insert into t1 values (3, 2);
+insert into t1 values (5, 2);
+commit;
+connection con1;
+# No ER_LOCK_DEADLOCK; all rows returned
+pk	x
+1	2
+2	2
+3	2
+4	2
+5	2
+connection default;
+disconnect con1;
+drop table t1;
+# C) Same as A) but reverse direction
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (3, 0);
+start transaction;
+update t1 set x= 3;
+connect  con1,localhost,root,,test;
+select * from t1 order by pk desc for update;
+connection default;
+insert into t1 values (1, 3);
+insert into t1 values (2, 3);
+insert into t1 values (4, 3);
+insert into t1 values (5, 3);
+commit;
+connection con1;
+# No ER_LOCK_DEADLOCK; all rows returned
+pk	x
+5	3
+4	3
+3	3
+2	3
+1	3
+connection default;
+disconnect con1;
+drop table t1;
+# D) Same as B) but reverse direction
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (2, 0);
+insert into t1 values (5, 4);
+start transaction;
+update t1 set x= 4 where pk < 3;
+connect  con1,localhost,root,,test;
+select * from t1 order by pk desc for update;
+connection default;
+insert into t1 values (1, 4);
+insert into t1 values (3, 4);
+insert into t1 values (4, 4);
+commit;
+connection con1;
+# No ER_LOCK_DEADLOCK; all rows returned
+pk	x
+5	4
+4	4
+3	4
+2	4
+1	4
+connection default;
+disconnect con1;
+drop table t1;
+# E) Same as A) but UPDATE instead SELECT
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (3, 0);
+start transaction;
+update t1 set x= 5;
+connect  con1, localhost, root,, test;
+update t1 set x= x + 10;
+connection default;
+insert into t1 values (1, 5);
+insert into t1 values (2, 5);
+insert into t1 values (4, 5);
+insert into t1 values (5, 5);
+commit;
+connection con1;
+# No ER_LOCK_DEADLOCK; all rows returned
+connection default;
+select * from t1;
+pk	x
+1	15
+2	15
+3	15
+4	15
+5	15
+disconnect con1;
+drop table t1;
+# F) Same as B) but UPDATE instead SELECT
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (1, 6);
+insert into t1 values (4, 0);
+start transaction;
+update t1 set x= 6 where pk > 3;
+connect  con1,localhost,root,,test;
+update t1 set x= x + 10;
+connection default;
+insert into t1 values (2, 6);
+insert into t1 values (3, 6);
+insert into t1 values (5, 6);
+commit;
+connection con1;
+# No ER_LOCK_DEADLOCK; all rows returned
+connection default;
+select * from t1;
+pk	x
+1	16
+2	16
+3	16
+4	16
+5	16
+disconnect con1;
+drop table t1;
+#
+# MDEV-10962 Deadlock with 3 concurrent DELETEs by unique key
+#
+create table t1 (a int unique) engine innodb;
+insert into t1 values (1);
+connect  con1, localhost, root,, test;
+connect  con2, localhost, root,, test;
+connect  con3, localhost, root,, test;
+connection con1;
+delete from t1 where a = 1;
+connection con2;
+delete from t1 where a = 1;
+connection con3;
+delete from t1 where a = 1;
+connection con1;
+connection con2;
+connection con3;
+connection default;
+disconnect con1;
+disconnect con2;
+disconnect con3;
+drop table t1;

mysql-test/suite/innodb/t/innodb-lock.opt

+--enable-plugin-innodb-lock-waits

mysql-test/suite/innodb/t/innodb-lock.test

@@ -219,3 +219,175 @@ reap;
 disconnect con1;
 connection default;
 DROP TABLE t1, t2, t3;
+
+--echo #
+--echo # MDEV-18706 ER_LOCK_DEADLOCK on concurrent read and insert into already locked gap
+--echo #
+--echo # A) Def inserts between infimum and Def-locked record, between Def-locked record and supremum
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (3, 0);
+start transaction;
+# 1. Def: X-lock "3" as "next key" (row + gap)
+update t1 set x= 1;
+connect (con1, localhost, root,, test);
+# 2. Con1: Block on "3"
+send select * from t1 for update;
+connection default;
+--let $wait_condition= select count(*) from information_schema.innodb_lock_waits
+--source include/wait_condition.inc
+# 3. Def: Insert before and after "3"
+insert into t1 values (1, 1);
+insert into t1 values (2, 1);
+insert into t1 values (4, 1);
+insert into t1 values (5, 1);
+# 4. Def: Unlock "3"
+commit;
+connection con1;
+# 5. Con1: Continue SELECT
+--echo # No ER_LOCK_DEADLOCK; all rows returned
+reap;
+connection default;
+disconnect con1;
+drop table t1;
+
+--echo # B) Def inserts between Con1-locked and Def-locked record
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (1, 2);
+insert into t1 values (4, 0);
+start transaction;
+# 1. Def: X-lock "4" as "next key" (row + gap)
+update t1 set x= 2 where pk > 3;
+connect (con1,localhost,root,,test);
+# 2. Con1: X-lock "1", block on "4"
+send select * from t1 for update;
+connection default;
+--let $wait_condition= select count(*) from information_schema.innodb_lock_waits
+--source include/wait_condition.inc
+# 3. Def: Insert before and after "4"
+# Note: we cannot INSERT before "1" because Con1 already X-locked it
+insert into t1 values (2, 2);
+insert into t1 values (3, 2);
+insert into t1 values (5, 2);
+commit;
+connection con1;
+--echo # No ER_LOCK_DEADLOCK; all rows returned
+reap;
+connection default;
+disconnect con1;
+drop table t1;
+
+--echo # C) Same as A) but reverse direction
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (3, 0);
+start transaction;
+update t1 set x= 3;
+connect (con1,localhost,root,,test);
+send select * from t1 order by pk desc for update;
+connection default;
+--let $wait_condition= select count(*) from information_schema.innodb_lock_waits
+--source include/wait_condition.inc
+insert into t1 values (1, 3);
+insert into t1 values (2, 3);
+insert into t1 values (4, 3);
+insert into t1 values (5, 3);
+commit;
+connection con1;
+--echo # No ER_LOCK_DEADLOCK; all rows returned
+reap;
+connection default;
+disconnect con1;
+drop table t1;
+
+--echo # D) Same as B) but reverse direction
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (2, 0);
+insert into t1 values (5, 4);
+start transaction;
+update t1 set x= 4 where pk < 3;
+connect (con1,localhost,root,,test);
+send select * from t1 order by pk desc for update;
+connection default;
+--let $wait_condition= select count(*) from information_schema.innodb_lock_waits
+--source include/wait_condition.inc
+insert into t1 values (1, 4);
+insert into t1 values (3, 4);
+insert into t1 values (4, 4);
+commit;
+connection con1;
+--echo # No ER_LOCK_DEADLOCK; all rows returned
+reap;
+connection default;
+disconnect con1;
+drop table t1;
+
+--echo # E) Same as A) but UPDATE instead SELECT
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (3, 0);
+start transaction;
+update t1 set x= 5;
+connect (con1, localhost, root,, test);
+send update t1 set x= x + 10;
+connection default;
+--let $wait_condition= select count(*) from information_schema.innodb_lock_waits
+--source include/wait_condition.inc
+insert into t1 values (1, 5);
+insert into t1 values (2, 5);
+insert into t1 values (4, 5);
+insert into t1 values (5, 5);
+commit;
+connection con1;
+--echo # No ER_LOCK_DEADLOCK; all rows returned
+reap;
+connection default;
+select * from t1;
+disconnect con1;
+drop table t1;
+
+--echo # F) Same as B) but UPDATE instead SELECT
+create or replace table t1 (pk int primary key, x int) engine innodb;
+insert into t1 values (1, 6);
+insert into t1 values (4, 0);
+start transaction;
+update t1 set x= 6 where pk > 3;
+connect (con1,localhost,root,,test);
+send update t1 set x= x + 10;
+connection default;
+--let $wait_condition= select count(*) from information_schema.innodb_lock_waits
+--source include/wait_condition.inc
+insert into t1 values (2, 6);
+insert into t1 values (3, 6);
+insert into t1 values (5, 6);
+commit;
+connection con1;
+--echo # No ER_LOCK_DEADLOCK; all rows returned
+reap;
+connection default;
+select * from t1;
+disconnect con1;
+drop table t1;
+
+--echo #
+--echo # MDEV-10962 Deadlock with 3 concurrent DELETEs by unique key
+--echo #
+create table t1 (a int unique) engine innodb;
+insert into t1 values (1);
+connect (con1, localhost, root,, test);
+connect (con2, localhost, root,, test);
+connect (con3, localhost, root,, test);
+connection con1;
+send delete from t1 where a = 1;
+connection con2;
+send delete from t1 where a = 1;
+connection con3;
+send delete from t1 where a = 1;
+connection con1;
+reap;
+connection con2;
+reap;
+connection con3;
+reap;
+connection default;
+disconnect con1;
+disconnect con2;
+disconnect con3;
+drop table t1;

mysql-test/t/xa.test

@@ -340,9 +340,7 @@ DROP TABLE t1;
 
 
 --echo #
---echo # Bug#12352846 - TRANS_XA_START(THD*):
---echo #                ASSERTION THD->TRANSACTION.XID_STATE.XID.IS_NULL()
---echo #                FAILED 
+--echo # MDEV-10962 Deadlock with 3 concurrent DELETEs by unique key
 --echo #
 
 --disable_warnings
@@ -369,20 +367,17 @@ let $wait_condition=
 --source include/wait_condition.inc
 --sleep 0.1
 DELETE FROM t1;
+COMMIT;
 
 --connection con2
 --echo # Reaping: INSERT INTO t2 SELECT a FROM t1
---error ER_LOCK_DEADLOCK
 --reap
---error ER_XA_RBDEADLOCK
+XA END 'xid1';
+XA PREPARE 'xid1';
 XA COMMIT 'xid1';
 
-connection default;
-
-COMMIT;
-
 connection con2;
-# This caused the assert to be triggered
+# This caused the assert to be triggered (Bug#12352846)
 XA START 'xid1';
 
 XA END 'xid1';

storage/innobase/handler/ha_innodb.cc

@@ -19684,6 +19684,7 @@ wsrep_innobase_kill_one_trx(
 
 			if (wait_lock) {
 				WSREP_DEBUG("canceling wait lock");
+				DBUG_LOG("ib_lock", VICTIM(victim_trx) << *wait_lock);
 				victim_trx->lock.was_chosen_as_deadlock_victim= TRUE;
 				lock_cancel_waiting_and_release(wait_lock);
 			}

storage/innobase/include/lock0lock.h

@@ -1050,4 +1050,28 @@ lock_get_info(
 
 #include "lock0lock.ic"
 
+/** The global output operator is overloaded to conveniently
+print the lock_table_t object into the given output stream.
+@param[in,out]	out	the output stream
+@param[in]	lock	the table lock
+@return the given output stream */
+inline
+std::ostream&
+operator<<(std::ostream& out, const lock_table_t& lock)
+{
+	return(lock.print(out));
+}
+
+/** The global output operator is overloaded to conveniently
+print the ib_lock_t object into the given output stream.
+@param[in,out]	out	the output stream
+@param[in]	lock	the record lock
+@return the given output stream */
+inline
+std::ostream&
+operator<<(std::ostream& out, const ib_lock_t& lock)
+{
+	return(lock.print(out));
+}
+
 #endif

storage/innobase/include/lock0priv.h

@@ -48,56 +48,21 @@ those functions in lock/ */
 inline
 std::ostream& lock_table_t::print(std::ostream& out) const
 {
-	out << "[lock_table_t: name=" << table->name << "]";
+	out << "[" << table->name.m_name << "]";
 	return(out);
 }
 
-/** The global output operator is overloaded to conveniently
-print the lock_table_t object into the given output stream.
-@param[in,out]	out	the output stream
-@param[in]	lock	the table lock
-@return the given output stream */
-inline
-std::ostream&
-operator<<(std::ostream& out, const lock_table_t& lock)
-{
-	return(lock.print(out));
-}
-
-/** Convert the member 'type_mode' into a human readable string.
-@return human readable string */
-inline
-std::string
-ib_lock_t::type_mode_string() const
-{
-	std::ostringstream sout;
-	sout << type_string();
-	sout << " | " << lock_mode_string(mode());
-
-	if (is_record_not_gap()) {
-		sout << " | LOCK_REC_NOT_GAP";
-	}
-
-	if (is_waiting()) {
-		sout << " | LOCK_WAIT";
-	}
-
-	if (is_gap()) {
-		sout << " | LOCK_GAP";
-	}
-
-	if (is_insert_intention()) {
-		sout << " | LOCK_INSERT_INTENTION";
-	}
-	return(sout.str());
-}
-
 inline
 std::ostream&
 ib_lock_t::print(std::ostream& out) const
 {
-	out << "[lock_t: type_mode=" << type_mode << "("
-		<< type_mode_string() << ")";
+	out << "[trx=" << trx << "(" << trx->lock.trx_locks.count
+		<< ":" << trx->lock.table_locks.size() << "), ";
+	if (index) {
+		out << "index=" << index << "("
+			<< (index->is_primary() ? "#" : index->name()) << "), ";
+	}
+	out << "type_mode=" << type_mode << "=" << type_mode_string() << " ";
 
 	if (is_record_lock()) {
 		out << un_member.rec_lock;
@@ -109,17 +74,6 @@ ib_lock_t::print(std::ostream& out) const
 	return(out);
 }
 
-inline
-std::ostream&
-operator<<(std::ostream& out, const ib_lock_t& lock)
-{
-	return(lock.print(out));
-}
-
-#ifdef UNIV_DEBUG
-extern ibool	lock_print_waits;
-#endif /* UNIV_DEBUG */
-
 /** Restricts the length of search we will do in the waits-for
 graph of transactions */
 static const ulint	LOCK_MAX_N_STEPS_IN_DEADLOCK_CHECK = 1000000;
@@ -594,6 +548,15 @@ lock_rec_get_first(
 	const buf_block_t*	block,	/*!< in: block containing the record */
 	ulint			heap_no);/*!< in: heap number of the record */
 
+/*********************************************************************//**
+Gets the mode from type_mode.
+@return mode */
+UNIV_INLINE
+enum lock_mode
+lock_get_mode(
+/*==========*/
+	const ib_uint32_t type_mode);
+
 /*********************************************************************//**
 Gets the mode of a lock.
 @return mode */
@@ -671,6 +634,7 @@ inline void lock_set_lock_and_trx_wait(lock_t* lock, trx_t* trx)
 
 	trx->lock.wait_lock = lock;
 	lock->type_mode |= LOCK_WAIT;
+	DBUG_LOG("ib_lock", "+WAIT("<< lock << ") " << *lock);
 }
 
 /** Reset the wait status of a lock.
@@ -683,6 +647,26 @@ inline void lock_reset_lock_and_trx_wait(lock_t* lock)
 	      || lock->trx->lock.wait_lock == lock);
 	lock->trx->lock.wait_lock = NULL;
 	lock->type_mode &= ~LOCK_WAIT;
+	DBUG_LOG("ib_lock", "-WAIT("<< lock << ") " << *lock);
+}
+
+inline
+bool ib_lock_t::is_stronger(ulint precise_mode, ulint heap_no, const trx_t* t) const
+{
+	ut_ad(is_record_lock());
+	return trx == t
+	    && !is_waiting()
+	    && !is_insert_intention()
+	    && (!is_record_not_gap()
+		|| (precise_mode & LOCK_REC_NOT_GAP) /* only record */
+		|| heap_no == PAGE_HEAP_NO_SUPREMUM)
+	    && (!is_gap()
+		|| (precise_mode & LOCK_GAP)         /* only gap */
+		|| heap_no == PAGE_HEAP_NO_SUPREMUM)
+	    && lock_mode_stronger_or_eq(
+		mode(),
+		static_cast<lock_mode>(
+			precise_mode & LOCK_MODE_MASK));
 }
 
 #include "lock0priv.ic"

storage/innobase/include/lock0priv.ic

@@ -286,6 +286,18 @@ lock_rec_get_next_on_page_const(
 	return(NULL);
 }
 
+/*********************************************************************//**
+Gets the mode from type_mode.
+@return mode */
+UNIV_INLINE
+enum lock_mode
+lock_get_mode(
+/*==========*/
+	const ib_uint32_t type_mode)
+{
+	return(static_cast<enum lock_mode>(type_mode & LOCK_MODE_MASK));
+}
+
 /*********************************************************************//**
 Gets the mode of a lock.
 @return mode */
@@ -297,7 +309,7 @@ lock_get_mode(
 {
 	ut_ad(lock);
 
-	return(static_cast<enum lock_mode>(lock->type_mode & LOCK_MODE_MASK));
+	return(lock_get_mode(lock->type_mode));
 }
 
 /*********************************************************************//**

storage/innobase/include/lock0types.h

@@ -57,19 +57,19 @@ const char* lock_mode_string(enum lock_mode mode)
 {
 	switch (mode) {
 	case LOCK_IS:
-		return("LOCK_IS");
+		return("IS");
 	case LOCK_IX:
-		return("LOCK_IX");
+		return("IX");
 	case LOCK_S:
-		return("LOCK_S");
+		return("S");
 	case LOCK_X:
-		return("LOCK_X");
+		return("X");
 	case LOCK_AUTO_INC:
-		return("LOCK_AUTO_INC");
+		return("AUTO_INC");
 	case LOCK_NONE:
-		return("LOCK_NONE");
+		return("NONE");
 	case LOCK_NONE_UNSET:
-		return("LOCK_NONE_UNSET");
+		return("NONE_UNSET");
 	default:
 		ut_error;
 	}
@@ -109,7 +109,7 @@ struct lock_rec_t {
 inline
 std::ostream& lock_rec_t::print(std::ostream& out) const
 {
-	out << "[lock_rec_t: space=" << space << ", page_no=" << page_no
+	out << "[space=" << space << ", page_no=" << page_no
 		<< ", n_bits=" << n_bits << "]";
 	return(out);
 }
@@ -176,6 +176,51 @@ operator<<(std::ostream& out, const lock_rec_t& lock)
 #endif
 /* @} */
 
+inline
+const char*
+type_string(ulint type_mode)
+{
+	switch (type_mode & LOCK_TYPE_MASK) {
+	case LOCK_REC:
+		return("REC");
+	case LOCK_TABLE:
+		return("TABLE");
+	default:
+		ut_error;
+	}
+}
+
+/** Convert 'type_mode' into a human readable string.
+@return human readable string */
+inline
+std::string
+type_mode_string(ulint type_mode)
+{
+	std::ostringstream sout;
+	lock_mode mode = static_cast<enum lock_mode>(type_mode & LOCK_MODE_MASK);
+	if (type_mode & LOCK_TYPE_MASK) {
+		sout << type_string(type_mode) << "|";
+	}
+	sout << lock_mode_string(mode);
+
+	if (type_mode & LOCK_REC_NOT_GAP) {
+		sout << "|REC_NOT_GAP";
+	}
+
+	if (type_mode & LOCK_WAIT) {
+		sout << "|WAIT";
+	}
+
+	if (type_mode & LOCK_GAP) {
+		sout << "|GAP";
+	}
+
+	if (type_mode & LOCK_INSERT_INTENTION) {
+		sout << "|INSERT_INTENTION";
+	}
+	return(sout.str());
+}
+
 /** Lock struct; protected by lock_sys->mutex */
 struct ib_lock_t
 {
@@ -237,7 +282,10 @@ struct ib_lock_t
 		return(type_mode & LOCK_INSERT_INTENTION);
 	}
 
-	ulint type() const {
+	bool is_stronger(ulint precise_mode, ulint heap_no, const trx_t* t) const;
+
+	ulint type() const
+	{
 		return(type_mode & LOCK_TYPE_MASK);
 	}
 
@@ -251,23 +299,98 @@ struct ib_lock_t
 	@return the given output stream. */
 	std::ostream& print(std::ostream& out) const;
 
-	/** Convert the member 'type_mode' into a human readable string.
-	@return human readable string */
-	std::string type_mode_string() const;
+	std::string type_mode_string() const
+	{
+		return ::type_mode_string(type_mode);
+	}
 
 	const char* type_string() const
 	{
-		switch (type_mode & LOCK_TYPE_MASK) {
-		case LOCK_REC:
-			return("LOCK_REC");
-		case LOCK_TABLE:
-			return("LOCK_TABLE");
-		default:
-			ut_error;
-		}
+		return ::type_string(type_mode);
 	}
 };
 
 typedef UT_LIST_BASE_NODE_T(ib_lock_t) trx_lock_list_t;
 
+#ifndef DBUG_OFF
+/* Classes used to catch various locking situations in code */
+
+struct ADD /* add lock */
+{
+	const lock_t *lock;
+	ADD(const lock_t *l) : lock(l)
+	{
+	}
+};
+
+inline
+std::ostream&
+operator<<(std::ostream& out, const ADD& a)
+{
+	out << "ADD(" << a.lock << ") ";
+	return out;
+}
+
+struct VICTIM /* deadlock victim */
+{
+	const trx_t *trx;
+	bool set;
+	VICTIM(const trx_t *t, bool s = true) : trx(t), set(s)
+	{
+	}
+};
+
+inline
+std::ostream&
+operator<<(std::ostream& out, const VICTIM& v)
+{
+	out << (v.set ? '+' : '-') << "VICTIM(trx=" << v.trx << ") ";
+	return out;
+}
+
+struct WEAKER /* precise_mode is weaker than existing lock (of same trx) */
+{
+	ulint precise_mode;
+	lock_t *lock;
+	WEAKER(ulint m, lock_t *l) : precise_mode(m), lock(l)
+	{
+	}
+};
+
+inline
+std::ostream&
+operator<<(std::ostream& out, const WEAKER& w)
+{
+	out << "WEAKER(" << type_mode_string(w.precise_mode) << ", " << w.lock << ") ";
+	w.lock->print(out);
+	out << " ";
+	return out;
+}
+
+struct CONFLICTS /* precise_mode conflicts (or doesn't) with any existing locks */
+{
+	const trx_t *trx;
+	ulint precise_mode;
+	const lock_t *conflict;
+	CONFLICTS(const trx_t *t, ulint m, const lock_t *c) :
+		trx(t), precise_mode(m), conflict(c)
+	{
+	}
+};
+
+inline
+std::ostream&
+operator<<(std::ostream& out, const CONFLICTS& c)
+{
+	out << (c.conflict ? "CONFLICTS(trx=" : "NO_CONFLICTS(trx=")
+		<< c.trx << ", " << type_mode_string(c.precise_mode) << ", "
+		<< c.conflict  << ") ";
+	if (c.conflict) {
+		c.conflict->print(out);
+		out << " ";
+	}
+	return out;
+}
+
+#endif /* !DBUG_OFF */
 #endif /* lock0types_h */

storage/innobase/lock/lock0lock.cc

@@ -1004,6 +1004,7 @@ lock_rec_has_expl(
 {
 	lock_t*	lock;
 
+	DBUG_ENTER("lock_rec_has_expl");
 	ut_ad(lock_mutex_own());
 	ut_ad((precise_mode & LOCK_MODE_MASK) == LOCK_S
 	      || (precise_mode & LOCK_MODE_MASK) == LOCK_X);
@@ -1013,25 +1014,14 @@ lock_rec_has_expl(
 	     lock != NULL;
 	     lock = lock_rec_get_next(heap_no, lock)) {
 
-		if (lock->trx == trx
-		    && !lock_rec_get_insert_intention(lock)
-		    && lock_mode_stronger_or_eq(
-			    lock_get_mode(lock),
-			    static_cast<lock_mode>(
-				    precise_mode & LOCK_MODE_MASK))
-		    && !lock_get_wait(lock)
-		    && (!lock_rec_get_rec_not_gap(lock)
-			|| (precise_mode & LOCK_REC_NOT_GAP)
-			|| heap_no == PAGE_HEAP_NO_SUPREMUM)
-		    && (!lock_rec_get_gap(lock)
-			|| (precise_mode & LOCK_GAP)
-			|| heap_no == PAGE_HEAP_NO_SUPREMUM)) {
+		if (lock->is_stronger(precise_mode, heap_no, trx)) {
 
-			return(lock);
+			DBUG_LOG("ib_lock", WEAKER(precise_mode, lock));
+			DBUG_RETURN(lock);
 		}
 	}
 
-	return(NULL);
+	DBUG_RETURN(NULL);
 }
 
 #ifdef UNIV_DEBUG
@@ -1156,39 +1146,65 @@ lock_rec_other_has_conflicting(
 /*===========================*/
 	ulint			mode,	/*!< in: LOCK_S or LOCK_X,
 					possibly ORed to LOCK_GAP or
-					LOC_REC_NOT_GAP,
+					LOCK_REC_NOT_GAP,
 					LOCK_INSERT_INTENTION */
 	const buf_block_t*	block,	/*!< in: buffer block containing
 					the record */
 	ulint			heap_no,/*!< in: heap number of the record */
 	const trx_t*		trx)	/*!< in: our transaction */
 {
-	lock_t*		lock;
+	lock_t*		conflict = NULL;
+	bool		skip_waiting = false;
 
+	DBUG_ENTER("lock_rec_other_has_conflicting");
 	ut_ad(lock_mutex_own());
 
 	bool	is_supremum = (heap_no == PAGE_HEAP_NO_SUPREMUM);
 
-	for (lock = lock_rec_get_first(lock_sys->rec_hash, block, heap_no);
+	for (lock_t* lock = lock_rec_get_first(lock_sys->rec_hash, block, heap_no);
 	     lock != NULL;
 	     lock = lock_rec_get_next(heap_no, lock)) {
 
-		if (lock_rec_has_to_wait(true, trx, mode, lock, is_supremum)) {
+		if (skip_waiting && lock->is_waiting()) {
+			continue;
+		}
+
+		/* If current trx already acquired a lock not weaker covering
+		same types then we don't have to wait for any locks. */
+		if (lock->is_stronger(mode, heap_no, trx)) {
+
+			DBUG_LOG("ib_lock", CONFLICTS(trx, mode, NULL)
+				 << "because: " << WEAKER(mode, lock)) ;
+			DBUG_RETURN(NULL);
+		} else if (lock->trx == trx && !lock->is_waiting()) {
+			if (conflict && conflict->is_waiting()) {
+				conflict = NULL;
+			}
+			skip_waiting = true;
+		} else if (lock_rec_has_to_wait(true, trx, mode, lock,
+						is_supremum)) {
+
+			if (!conflict || (conflict->is_waiting()
+					  && !lock->is_waiting())) {
+
+				conflict = lock;
+			}
+		}
+	}
+
 #ifdef WITH_WSREP
-			if (trx->is_wsrep()) {
-				trx_mutex_enter(lock->trx);
+	if (conflict && trx->is_wsrep()) {
+		trx_mutex_enter(conflict->trx);
 		/* Below function will roll back either trx
 		or lock->trx depending on priority of the
 		transaction. */
-				wsrep_kill_victim(const_cast<trx_t*>(trx), lock);
-				trx_mutex_exit(lock->trx);
+		wsrep_kill_victim(const_cast<trx_t*>(trx), conflict);
+		trx_mutex_exit(conflict->trx);
 	}
 #endif /* WITH_WSREP */
-			return(lock);
-		}
-	}
 
-	return(NULL);
+	DBUG_LOG("ib_lock", CONFLICTS(trx, mode, conflict));
+	DBUG_RETURN(conflict);
 }
 
 /*********************************************************************//**
@@ -1464,6 +1480,7 @@ lock_rec_create_low(
 		lock->un_member.rec_lock.n_bits = 8;
  	}
 	lock_rec_bitmap_reset(lock);
+	DBUG_LOG("ib_lock", ADD(lock) << *lock);
 	lock_rec_set_nth_bit(lock, heap_no);
 	index->table->n_rec_locks++;
 	ut_ad(index->table->get_ref_count() > 0 || !index->table->can_be_evicted);
@@ -1492,7 +1509,7 @@ lock_rec_create_low(
 		 */
 		trx_mutex_enter(c_lock->trx);
 		if (c_lock->trx->lock.que_state == TRX_QUE_LOCK_WAIT) {
-
+			DBUG_LOG("ib_lock", VICTIM(c_lock->trx) << *c_lock);
 			c_lock->trx->lock.was_chosen_as_deadlock_victim = TRUE;
 
 			if (UNIV_UNLIKELY(wsrep_debug)) {
@@ -3640,6 +3657,7 @@ lock_table_create(
 		trx_mutex_enter(c_lock->trx);
 
 		if (c_lock->trx->lock.que_state == TRX_QUE_LOCK_WAIT) {
+			DBUG_LOG("ib_lock", VICTIM(c_lock->trx) << *c_lock);
 			c_lock->trx->lock.was_chosen_as_deadlock_victim = TRUE;
 
 			if (UNIV_UNLIKELY(wsrep_debug)) {
@@ -3666,6 +3684,7 @@ lock_table_create(
 		lock_set_lock_and_trx_wait(lock, trx);
 	}
 
+	DBUG_LOG("ib_lock", ADD(lock) << *lock);
 	lock->trx->lock.table_locks.push_back(lock);
 
 	MONITOR_INC(MONITOR_TABLELOCK_CREATED);
@@ -3801,6 +3820,8 @@ lock_table_remove_low(
 	UT_LIST_REMOVE(trx->lock.trx_locks, lock);
 	ut_list_remove(table->locks, lock, TableLockGetNode());
 
+	DBUG_LOG("ib_lock", "DEL("<< lock << ") " << *lock);
+
 	MONITOR_INC(MONITOR_TABLELOCK_REMOVED);
 	MONITOR_DEC(MONITOR_NUM_TABLELOCK);
 }
@@ -3846,7 +3867,8 @@ lock_table_enqueue_waiting(
 	}
 
 #ifdef WITH_WSREP
-	if (trx->is_wsrep() && trx->lock.was_chosen_as_deadlock_victim) {
+	if (trx->lock.was_chosen_as_deadlock_victim && trx->is_wsrep()) {
+		DBUG_LOG("ib_lock", "DEADLOCK(" << trx << ") ");
 		return(DB_DEADLOCK);
 	}
 #endif /* WITH_WSREP */
@@ -3869,6 +3891,7 @@ lock_table_enqueue_waiting(
 		lock_table_remove_low(lock);
 		lock_reset_lock_and_trx_wait(lock);
 
+		DBUG_LOG("ib_lock", "DEADLOCK(" << trx << ") ");
 		return(DB_DEADLOCK);
 
 	} else if (trx->lock.wait_lock == NULL) {
@@ -3881,6 +3904,7 @@ lock_table_enqueue_waiting(
 	trx->lock.que_state = TRX_QUE_LOCK_WAIT;
 
 	trx->lock.wait_started = time(NULL);
+	DBUG_LOG("ib_lock", VICTIM(trx, false));
 	trx->lock.was_chosen_as_deadlock_victim = false;
 
 	ut_a(que_thr_stop(thr));
@@ -5762,6 +5786,7 @@ lock_rec_convert_impl_to_expl_for_trx(
 	if (!trx_state_eq(trx, TRX_STATE_COMMITTED_IN_MEMORY)
 	    && !lock_rec_has_expl(LOCK_X | LOCK_REC_NOT_GAP,
 				  block, heap_no, trx)) {
+		DBUG_LOG("ib_lock", "IMPL_TO_EXPL(trx=" << trx << ")");
 		lock_rec_add_to_queue(LOCK_REC | LOCK_X | LOCK_REC_NOT_GAP,
 				      block, heap_no, index, trx, true);
 	}
@@ -7182,6 +7207,7 @@ DeadlockChecker::trx_rollback()
 
 	trx_mutex_enter(trx);
 
+	DBUG_LOG("ib_lock", VICTIM(trx));
 	trx->lock.was_chosen_as_deadlock_victim = true;
 
 	lock_cancel_waiting_and_release(trx->lock.wait_lock);

storage/innobase/lock/lock0wait.cc

@@ -262,6 +262,7 @@ lock_wait_suspend_thread(
 		if (trx->lock.was_chosen_as_deadlock_victim) {
 
 			trx->error_state = DB_DEADLOCK;
+			DBUG_LOG("ib_lock", "DEADLOCK(" << trx << ") ");
 			trx->lock.was_chosen_as_deadlock_victim = false;
 		}
 
@@ -441,6 +442,7 @@ lock_wait_release_thread_if_suspended(
 		if (trx->lock.was_chosen_as_deadlock_victim) {
 
 			trx->error_state = DB_DEADLOCK;
+			DBUG_LOG("ib_lock", "DEADLOCK(" << trx << ") ");
 			trx->lock.was_chosen_as_deadlock_victim = false;
 		}
 

storage/innobase/row/row0sel.cc

@@ -5770,6 +5770,13 @@ row_search_mvcc(
 
 lock_wait_or_error:
 	if (!dict_index_is_spatial(index)) {
+		/* Locked gap may be filled with inserted records.
+		Make sure we don't miss them. */
+		if (moves_up) {
+			btr_pcur_move_to_prev(pcur, &mtr);
+		} else {
+			btr_pcur_move_to_next(pcur, &mtr);
+		}
 		btr_pcur_store_position(pcur, &mtr);
 	}
 page_read_error:
@@ -5812,6 +5819,16 @@ row_search_mvcc(
 			sel_restore_position_for_mysql(
 				&same_user_rec, BTR_SEARCH_LEAF, pcur,
 				moves_up, &mtr);
+			/* Counterpart of the stepping backward in lock_wait_or_error.
+			This is linked tight with that btr_pcur_store_position().
+			The jumps to page_read_error: and lock_table_wait: do not get here. */
+			if (same_user_rec) {
+				if (!moves_up) {
+					btr_pcur_move_to_prev(pcur, &mtr);
+				} else {
+					btr_pcur_move_to_next(pcur, &mtr);
+				}
+			}
 		}
 
 		if ((srv_locks_unsafe_for_binlog

storage/innobase/ut/ut0ut.cc

@@ -673,4 +673,39 @@ fatal_or_error::~fatal_or_error()
 
 } // namespace ib
 
+#ifndef DBUG_OFF
+static std::string dbug_str;
+
+template <class T>
+const char * dbug_print(T &obj)
+{
+	std::ostringstream os;
+	os.str("");
+	os.clear();
+	obj.print(os);
+	dbug_str = os.str();
+	return dbug_str.c_str();
+}
+
+const char * dbug_print(ib_lock_t *obj)
+{
+	return dbug_print(*obj);
+}
+
+const char * dbug_print(lock_rec_t *obj)
+{
+	return dbug_print(*obj);
+}
+
+const char * dbug_print(lock_table_t *obj)
+{
+	return dbug_print(*obj);
+}
+
+const char * dbug_print_lock_mode(ib_uint32_t type_mode)
+{
+	dbug_str = type_mode_string(type_mode);
+	return dbug_str.c_str();
+}
+#endif /* !DBUG_OFF */
 #endif /* !UNIV_INNOCHECKSUM */