Implement new type-of-operation-aware metadata locks.

Add a wait-for graph based deadlock detector to the
MDL subsystem.

Fixes bug #46272 "MySQL 5.4.4, new MDL: unnecessary deadlock" and
bug #37346 "innodb does not detect deadlock between update and
alter table".

The first bug manifested itself as an unwarranted abort of a
transaction with ER_LOCK_DEADLOCK error by a concurrent ALTER
statement, when this transaction tried to repeat use of a
table, which it has already used in a similar fashion before
ALTER started.

The second bug showed up as a deadlock between table-level
locks and InnoDB row locks, which was "detected" only after
innodb_lock_wait_timeout timeout.

A transaction would start using the table and modify a few
rows.
Then ALTER TABLE would come in, and start copying rows
into a temporary table. Eventually it would stumble on
the modified records and get blocked on a row lock.
The first transaction would try to do more updates, and get
blocked on thr_lock.c lock.
This situation of circular wait would only get resolved
by a timeout.

Both these bugs stemmed from inadequate solutions to the
problem of deadlocks occurring between different
locking subsystems.

In the first case we tried to avoid deadlocks between metadata
locking and table-level locking subsystems, when upgrading shared
metadata lock to exclusive one.
Transactions holding the shared lock on the table and waiting for
some table-level lock used to be aborted too aggressively.

We also allowed ALTER TABLE to start in presence of transactions
that modify the subject table. ALTER TABLE acquires
TL_WRITE_ALLOW_READ lock at start, and that block all writes
against the table (naturally, we don't want any writes to be lost
when switching the old and the new table). TL_WRITE_ALLOW_READ
lock, in turn, would block the started transaction on thr_lock.c
lock, should they do more updates. This, again, lead to the need
to abort such transactions.

The second bug occurred simply because we didn't have any
mechanism to detect deadlocks between the table-level locks
in thr_lock.c and row-level locks in InnoDB, other than
innodb_lock_wait_timeout.

This patch solves both these problems by moving lock conflicts
which are causing these deadlocks into the metadata locking
subsystem, thus making it possible to avoid or detect such
deadlocks inside MDL.

To do this we introduce new type-of-operation-aware metadata
locks, which allow MDL subsystem to know not only the fact that
transaction has used or is going to use some object but also what
kind of operation it has carried out or going to carry out on the
object.

This, along with the addition of a special kind of upgradable
metadata lock, allows ALTER TABLE to wait until all
transactions which has updated the table to go away.
This solves the second issue.
Another special type of upgradable metadata lock is acquired
by LOCK TABLE WRITE. This second lock type allows to solve the
first issue, since abortion of table-level locks in event of
DDL under LOCK TABLES becomes also unnecessary.

Below follows the list of incompatible changes introduced by
this patch:

- From now on, ALTER TABLE and CREATE/DROP TRIGGER SQL (i.e. those
  statements that acquire TL_WRITE_ALLOW_READ lock)
  wait for all transactions which has *updated* the table to
  complete.

- From now on, LOCK TABLES ... WRITE, REPAIR/OPTIMIZE TABLE
  (i.e. all statements which acquire TL_WRITE table-level lock) wait
  for all transaction which *updated or read* from the table
  to complete.
  As a consequence, innodb_table_locks=0 option no longer applies
  to LOCK TABLES ... WRITE.

- DROP DATABASE, DROP TABLE, RENAME TABLE no longer abort
  statements or transactions which use tables being dropped or
  renamed, and instead wait for these transactions to complete.

- Since LOCK TABLES WRITE now takes a special metadata lock,
  not compatible with with reads or writes against the subject table
  and transaction-wide, thr_lock.c deadlock avoidance algorithm
  that used to ensure absence of deadlocks between LOCK TABLES
  WRITE and other statements is no longer sufficient, even for
  MyISAM. The wait-for graph based deadlock detector of MDL
  subsystem may sometimes be necessary and is involved. This may
  lead to ER_LOCK_DEADLOCK error produced for multi-statement
  transactions even if these only use MyISAM:

  session 1:         session 2:
  begin;

  update t1 ...      lock table t2 write, t1 write;
                     -- gets a lock on t2, blocks on t1

  update t2 ...
  (ER_LOCK_DEADLOCK)

- Finally,  support of LOW_PRIORITY option for LOCK TABLES ... WRITE
  was abandoned.
  LOCK TABLE ... LOW_PRIORITY WRITE from now on has the same
  priority as the usual LOCK TABLE ... WRITE.
  SELECT HIGH PRIORITY no longer trumps LOCK TABLE ... WRITE  in
  the wait queue.

- We do not take upgradable metadata locks on implicitly
  locked tables. So if one has, say, a view v1 that uses
  table t1, and issues:
  LOCK TABLE v1 WRITE;
  FLUSH TABLE t1; -- (or just 'FLUSH TABLES'),
  an error is produced.
  In order to be able to perform DDL on a table under LOCK TABLES,
  the table must be locked explicitly in the LOCK TABLES list.

mysql-test/include/handler.inc:
  Adjusted test case to trigger an execution path on which bug 41110
  "crash with handler command when used concurrently with alter
  table" and bug 41112 "crash in mysql_ha_close_table/get_lock_data
  with alter table" were originally discovered. Left old test case
  which no longer triggers this execution path for the sake of
  coverage.
  Added test coverage for HANDLER SQL statements and type-aware
  metadata locks.
  Added a test for the global shared lock and HANDLER SQL.
  Updated tests to take into account that the old simple deadlock
  detection heuristics was replaced with a graph-based deadlock
  detector.
mysql-test/r/debug_sync.result:
  Updated results (see debug_sync.test).
mysql-test/r/handler_innodb.result:
  Updated results (see handler.inc test).
mysql-test/r/handler_myisam.result:
  Updated results (see handler.inc test).
mysql-test/r/innodb-lock.result:
  Updated results (see innodb-lock.test).
mysql-test/r/innodb_mysql_lock.result:
  Updated results (see innodb_mysql_lock.test).
mysql-test/r/lock.result:
  Updated results (see lock.test).
mysql-test/r/lock_multi.result:
  Updated results (see lock_multi.test).
mysql-test/r/lock_sync.result:
  Updated results (see lock_sync.test).
mysql-test/r/mdl_sync.result:
  Updated results (see mdl_sync.test).
mysql-test/r/sp-threads.result:
  SHOW PROCESSLIST output has changed due to the fact that waiting
  for LOCK TABLES WRITE now happens within metadata locking
  subsystem.
mysql-test/r/truncate_coverage.result:
  Updated results (see truncate_coverage.test).
mysql-test/suite/funcs_1/datadict/processlist_val.inc:
  SELECT FROM I_S.PROCESSLIST output has changed due to fact that
  waiting for LOCK TABLES WRITE now happens within metadata locking
  subsystem.
mysql-test/suite/funcs_1/r/processlist_val_no_prot.result:
  SELECT FROM I_S.PROCESSLIST output has changed due to fact that
  waiting for LOCK TABLES WRITE now happens within metadata locking
  subsystem.
mysql-test/suite/rpl/t/rpl_sp.test:
  Updated to a new SHOW PROCESSLIST state name.
mysql-test/t/debug_sync.test:
  Use LOCK TABLES READ instead of LOCK TABLES WRITE as the latter
  no longer allows to trigger execution path involving waiting on
  thr_lock.c lock and therefore reaching debug sync-point covered
  by this test.
mysql-test/t/innodb-lock.test:
  Adjusted test case to the fact that innodb_table_locks=0 option is
  no longer supported, since LOCK TABLES WRITE handles all its
  conflicts within MDL subsystem.
mysql-test/t/innodb_mysql_lock.test:
  Added test for bug #37346 "innodb does not detect deadlock between
  update and alter table".
mysql-test/t/lock.test:
  Added test coverage which checks the fact that we no longer support
  DDL under LOCK TABLES on tables which were locked implicitly.
  Adjusted existing test cases accordingly.
mysql-test/t/lock_multi.test:
  Added test for bug #46272 "MySQL 5.4.4, new MDL: unnecessary
  deadlock".  Adjusted other test cases to take into account the
  fact that waiting for LOCK TABLES ... WRITE now happens within MDL
  subsystem.
mysql-test/t/lock_sync.test:
  Since LOCK TABLES ... WRITE now takes SNRW metadata lock for
  tables locked explicitly we have to implicitly lock InnoDB tables
  (through view) to trigger the table-level lock conflict between
  TL_WRITE and TL_WRITE_ALLOW_WRITE.
mysql-test/t/mdl_sync.test:
  Added basic test coverage for type-of-operation-aware metadata
  locks. Also covered with tests some use cases involving HANDLER
  statements in which a deadlock could arise.
  Adjusted existing tests to take type-of-operation-aware MDL into
  account.
mysql-test/t/multi_update.test:
  Update to a new SHOW PROCESSLIST state name.
mysql-test/t/truncate_coverage.test:
  Adjusted test case after making LOCK TABLES WRITE to wait until
  transactions that use the table to be locked are completed.
  Updated to the changed name of DEBUG_SYNC point.
sql/handler.cc:
  Global read lock functionality has been
  moved into a class.
sql/lock.cc:
  Global read lock functionality has been
  moved into a class.
  Updated code to use the new MDL API.
sql/mdl.cc:
  Introduced new type-of-operation aware metadata locks.
  To do this:
  - Changed MDL_lock to use one list for waiting requests and one
    list for granted requests. For each list, added a bitmap
    that holds information what lock types a list contains.
    Added a helper class MDL_lock::List to manipulate with granted
    and waited lists while keeping the bitmaps in sync
    with list contents.
  - Changed lock-compatibility functions to use bitmaps that
    define compatibility.
  - Introduced a graph based deadlock detector inspired by
    waiting_threads.c from Maria implementation.
  - Now that we have a deadlock detector, and no longer have
    a global lock to protect individual lock objects, but rather
    use an rw lock per object, removed redundant code for upgrade,
    and the global read lock. Changed the MDL API to
    no longer require the caller to acquire the global
    intention exclusive lock by means of a separate method.
    Removed a few more methods that became redundant.
  - Removed deadlock detection heuristic, it has been made
    obsolete by the deadlock detector.
  - With operation-type-aware metadata locks, MDL subsystem has
    become aware of potential conflicts between DDL and open
    transactions. This made it possible to remove calls to
    mysql_abort_transactions_with_shared_lock() from acquisition
    paths for exclusive lock and lock upgrade. Now we can simply
    wait for these transactions to complete without fear of
    deadlock. Function mysql_lock_abort() has also become
    unnecessary for all conflicting cases except when a DDL
    conflicts with a connection that has an open HANDLER.
sql/mdl.h:
  Introduced new type-of-operation aware metadata locks.
  Introduced a graph based deadlock detector and supporting
  methods.
  Added comments.
  God rid of redundant API calls.
  Renamed m_lt_or_ha_sentinel to m_trans_sentinel,
  since now it guards the global read lock as well as
  LOCK TABLES and HANDLER locks.
sql/mysql_priv.h:
  Moved the global read lock functionality into a
  class.
  Added MYSQL_OPEN_FORCE_SHARED_MDL flag which forces
  open_tables() to take MDL_SHARED on tables instead of
  metadata locks specified in the parser. We use this to
  allow PREPARE run concurrently in presence of
  LOCK TABLES ... WRITE.
  Added signature for find_table_for_mdl_ugprade().
sql/set_var.cc:
  Global read lock functionality has been
  moved into a class.
sql/sp_head.cc:
  When creating TABLE_LIST elements for prelocking or
  system tables set the type of request for metadata
  lock according to the operation that will be performed
  on the table.
sql/sql_base.cc:
  - Updated code to use the new MDL API.
  - In order to avoid locks starvation we take upgradable
    locks all at once. As result implicitly locked tables no
    longer get an upgradable lock. Consequently DDL and FLUSH
    TABLES for such tables is prohibited.
    find_write_locked_table() was replaced by
    find_table_for_mdl_upgrade() function.
    open_table() was adjusted to return TABLE instance with
    upgradable ticket when necessary.
  - We no longer wait for all locks on OT_WAIT back off
    action -- only on the lock that caused the wait
    conflict. Moreover, now we distinguish cases when we
    have to wait due to conflict in MDL and old version
    of table in TDC.
  - Upate mysql_notify_threads_having_share_locks()
    to only abort thr_lock.c waits of threads that
    have open HANDLERs, since lock conflicts with only
    these threads now can lead to deadlocks not detectable
    by the MDL deadlock detector.
  - Remove mysql_abort_transactions_with_shared_locks()
    which is no longer needed.
sql/sql_class.cc:
  Global read lock functionality has been moved into a class.
  Re-arranged code in THD::cleanup() to simplify assert.
sql/sql_class.h:
  Introduced class to incapsulate global read lock
  functionality.
  Now sentinel in MDL subsystem guards the global read lock
  as well as LOCK TABLES and HANDLER locks. Adjusted code
  accordingly.
sql/sql_db.cc:
  Global read lock functionality has been moved into a class.
sql/sql_delete.cc:
  We no longer acquire upgradable metadata locks on tables
  which are locked by LOCK TABLES implicitly. As result
  TRUNCATE TABLE is no longer allowed for such tables.
  Updated code to use the new MDL API.
sql/sql_handler.cc:
  Inform MDL_context about presence of open HANDLERs.
  Since HANLDERs break MDL protocol by acquiring table-level
  lock while holding only S metadata lock on a table MDL
  subsystem should take special care about such contexts (Now
  this is the only case when mysql_lock_abort() is used).
sql/sql_parse.cc:
  Global read lock functionality has been moved into a class.
  Do not take upgradable metadata locks when opening tables
  for CREATE TABLE SELECT as it is not necessary and limits
  concurrency.
  When initializing TABLE_LIST objects before adding them
  to the table list set the type of request for metadata lock
  according to the operation that will be performed on the
  table.
  We no longer acquire upgradable metadata locks on tables
  which are locked by LOCK TABLES implicitly. As result FLUSH
  TABLES is no longer allowed for such tables.
sql/sql_prepare.cc:
  Use MYSQL_OPEN_FORCE_SHARED_MDL flag when opening
  tables during PREPARE. This allows PREPARE to run
  concurrently in presence of LOCK TABLES ... WRITE.
sql/sql_rename.cc:
  Global read lock functionality has been moved into a class.
sql/sql_show.cc:
  Updated code to use the new MDL API.
sql/sql_table.cc:
  Global read lock functionality has been moved into a class.
  We no longer acquire upgradable metadata locks on tables
  which are locked by LOCK TABLES implicitly. As result DROP
  TABLE is no longer allowed for such tables.
  Updated code to use the new MDL API.
sql/sql_trigger.cc:
  Global read lock functionality has been moved into a class.
  We no longer acquire upgradable metadata locks on tables
  which are locked by LOCK TABLES implicitly. As result
  CREATE/DROP TRIGGER is no longer allowed for such tables.
  Updated code to use the new MDL API.
sql/sql_view.cc:
  Global read lock functionality has been moved into a class.
  Fixed results of wrong merge that led to misuse of GLR API.
  CREATE VIEW statement is not a commit statement.
sql/table.cc:
  When resetting TABLE_LIST objects for PS or SP re-execution
  set the type of request for metadata lock according to the
  operation that will be performed on the table. Do the same
  in auxiliary function initializing metadata lock requests
  in a table list.
sql/table.h:
  When initializing TABLE_LIST objects set the type of request
  for metadata lock according to the operation that will be
  performed on the table.
sql/transaction.cc:
  Global read lock functionality has been moved into a class.

23 files changed, 1616 insertions, 1467 deletions

diff --git a/sql/handler.cc b/sql/handler.cc
index 66b0450d8d6..4becdaa3f4f 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1164,7 +1164,7 @@ int ha_commit_trans(THD *thd, bool all)
     rw_trans= is_real_trans && (rw_ha_count > 0);
 
     if (rw_trans &&
-        wait_if_global_read_lock(thd, 0, 0))
+        thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, FALSE))
     {
       ha_rollback_trans(thd, all);
       DBUG_RETURN(1);
@@ -1223,7 +1223,7 @@ int ha_commit_trans(THD *thd, bool all)
     RUN_HOOK(transaction, after_commit, (thd, FALSE));
 end:
     if (rw_trans)
-      start_waiting_global_read_lock(thd);
+      thd->global_read_lock.start_waiting_global_read_lock(thd);
   }
   /* Free resources and perform other cleanup even for 'empty' transactions. */
   else if (is_real_trans)
diff --git a/sql/lock.cc b/sql/lock.cc
index 9d794b07418..bd4068046ca 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -271,7 +271,7 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count,
 	Someone has issued LOCK ALL TABLES FOR READ and we want a write lock
 	Wait until the lock is gone
       */
-      if (wait_if_global_read_lock(thd, 1, 1))
+      if (thd->global_read_lock.wait_if_global_read_lock(thd, 1, 1))
       {
         /* Clear the lock type of all lock data to avoid reusage. */
         reset_lock_data_and_free(&sql_lock);
@@ -940,7 +940,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
    @note This function assumes that no metadata locks were acquired
          before calling it. Also it cannot be called while holding
          LOCK_open mutex. Both these invariants are enforced by asserts
-         in MDL_context::acquire_exclusive_locks().
+         in MDL_context::acquire_locks().
 
    @retval FALSE  Success.
    @retval TRUE   Failure (OOM or thread was killed).
@@ -962,14 +962,11 @@ bool lock_table_names(THD *thd, TABLE_LIST *table_list)
     mdl_requests.push_front(&lock_table->mdl_request);
   }
 
-  if (thd->mdl_context.acquire_global_intention_exclusive_lock(&global_request))
-    return 1;
+  mdl_requests.push_front(&global_request);
 
-  if (thd->mdl_context.acquire_exclusive_locks(&mdl_requests))
-  {
-    thd->mdl_context.release_lock(global_request.ticket);
+  if (thd->mdl_context.acquire_locks(&mdl_requests))
     return 1;
-  }
+
   return 0;
 }
 
@@ -1002,7 +999,7 @@ void unlock_table_names(THD *thd)
   This function assumes that no metadata locks were acquired
   before calling it. Additionally, it cannot be called while
   holding LOCK_open mutex. Both these invariants are enforced by
-  asserts in MDL_context::acquire_exclusive_locks().
+  asserts in MDL_context::acquire_locks().
   To avoid deadlocks, we do not try to obtain exclusive metadata
   locks in LOCK TABLES mode, since in this mode there may be
   other metadata locks already taken by the current connection,
@@ -1019,6 +1016,7 @@ bool lock_routine_name(THD *thd, bool is_function,
   MDL_key::enum_mdl_namespace mdl_type= (is_function ?
                                          MDL_key::FUNCTION :
                                          MDL_key::PROCEDURE);
+  MDL_request_list mdl_requests;
   MDL_request global_request;
   MDL_request mdl_request;
 
@@ -1035,14 +1033,11 @@ bool lock_routine_name(THD *thd, bool is_function,
   global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
   mdl_request.init(mdl_type, db, name, MDL_EXCLUSIVE);
 
-  if (thd->mdl_context.acquire_global_intention_exclusive_lock(&global_request))
-    return TRUE;
+  mdl_requests.push_front(&mdl_request);
+  mdl_requests.push_front(&global_request);
 
-  if (thd->mdl_context.acquire_exclusive_lock(&mdl_request))
-  {
-    thd->mdl_context.release_lock(global_request.ticket);
+  if (thd->mdl_context.acquire_locks(&mdl_requests))
     return TRUE;
-  }
 
   DEBUG_SYNC(thd, "after_wait_locked_pname");
   return FALSE;
@@ -1161,9 +1156,6 @@ volatile uint global_read_lock_blocks_commit=0;
 static volatile uint protect_against_global_read_lock=0;
 static volatile uint waiting_for_read_lock=0;
 
-#define GOT_GLOBAL_READ_LOCK               1
-#define MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT 2
-
 /**
   Take global read lock, wait if there is protection against lock.
 
@@ -1177,12 +1169,13 @@ static volatile uint waiting_for_read_lock=0;
   @retval True   Failure, thread was killed.
 */
 
-bool lock_global_read_lock(THD *thd)
+bool Global_read_lock::lock_global_read_lock(THD *thd)
 {
   DBUG_ENTER("lock_global_read_lock");
 
-  if (!thd->global_read_lock)
+  if (!m_state)
   {
+    MDL_request mdl_request;
     const char *old_message;
     const char *new_message= "Waiting to get readlock";
     (void) pthread_mutex_lock(&LOCK_global_read_lock);
@@ -1224,7 +1217,7 @@ bool lock_global_read_lock(THD *thd)
       thd->exit_cond(old_message);
       DBUG_RETURN(1);
     }
-    thd->global_read_lock= GOT_GLOBAL_READ_LOCK;
+    m_state= GRL_ACQUIRED;
     global_read_lock++;
     thd->exit_cond(old_message); // this unlocks LOCK_global_read_lock
     /*
@@ -1241,7 +1234,12 @@ bool lock_global_read_lock(THD *thd)
             redundancy between metadata locks, global read lock and DDL
             blocker (see WL#4399 and WL#4400).
     */
-    if (thd->mdl_context.acquire_global_shared_lock())
+
+    DBUG_ASSERT(! thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
+                                                 MDL_SHARED));
+    mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED);
+
+    if (thd->mdl_context.acquire_lock(&mdl_request))
     {
       /* Our thread was killed -- return back to initial state. */
       pthread_mutex_lock(&LOCK_global_read_lock);
@@ -1251,9 +1249,11 @@ bool lock_global_read_lock(THD *thd)
         pthread_cond_broadcast(&COND_global_read_lock);
       }
       pthread_mutex_unlock(&LOCK_global_read_lock);
-      thd->global_read_lock= 0;
+      m_state= GRL_NONE;
       DBUG_RETURN(1);
     }
+    thd->mdl_context.move_ticket_after_trans_sentinel(mdl_request.ticket);
+    m_mdl_global_shared_lock= mdl_request.ticket;
   }
   /*
     We DON'T set global_read_lock_blocks_commit now, it will be set after
@@ -1277,7 +1277,7 @@ bool lock_global_read_lock(THD *thd)
   @param thd    Reference to thread.
 */
 
-void unlock_global_read_lock(THD *thd)
+void Global_read_lock::unlock_global_read_lock(THD *thd)
 {
   uint tmp;
   DBUG_ENTER("unlock_global_read_lock");
@@ -1285,11 +1285,14 @@ void unlock_global_read_lock(THD *thd)
              ("global_read_lock: %u  global_read_lock_blocks_commit: %u",
               global_read_lock, global_read_lock_blocks_commit));
 
-  thd->mdl_context.release_global_shared_lock();
+  DBUG_ASSERT(m_mdl_global_shared_lock && m_state);
+
+  thd->mdl_context.release_lock(m_mdl_global_shared_lock);
+  m_mdl_global_shared_lock= NULL;
 
   pthread_mutex_lock(&LOCK_global_read_lock);
   tmp= --global_read_lock;
-  if (thd->global_read_lock == MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT)
+  if (m_state == GRL_ACQUIRED_AND_BLOCKS_COMMIT)
     --global_read_lock_blocks_commit;
   pthread_mutex_unlock(&LOCK_global_read_lock);
   /* Send the signal outside the mutex to avoid a context switch */
@@ -1298,7 +1301,7 @@ void unlock_global_read_lock(THD *thd)
     DBUG_PRINT("signal", ("Broadcasting COND_global_read_lock"));
     pthread_cond_broadcast(&COND_global_read_lock);
   }
-  thd->global_read_lock= 0;
+  m_state= GRL_NONE;
 
   DBUG_VOID_RETURN;
 }
@@ -1326,8 +1329,9 @@ void unlock_global_read_lock(THD *thd)
                    (is_not_commit ||                               \
                     global_read_lock_blocks_commit))
 
-bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
-                              bool is_not_commit)
+bool Global_read_lock::
+wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
+                         bool is_not_commit)
 {
   const char *UNINIT_VAR(old_message);
   bool result= 0, need_exit_cond;
@@ -1337,10 +1341,10 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
     If we already have protection against global read lock,
     just increment the counter.
   */
-  if (unlikely(thd->global_read_lock_protection > 0))
+  if (unlikely(m_protection_count > 0))
   {
     if (!abort_on_refresh)
-      thd->global_read_lock_protection++;
+      m_protection_count++;
     DBUG_RETURN(FALSE);
   }
   /*
@@ -1353,7 +1357,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
   (void) pthread_mutex_lock(&LOCK_global_read_lock);
   if ((need_exit_cond= must_wait))
   {
-    if (thd->global_read_lock)		// This thread had the read locks
+    if (m_state)		// This thread had the read locks
     {
       if (is_not_commit)
         my_message(ER_CANT_UPDATE_WITH_READLOCK,
@@ -1380,7 +1384,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
   }
   if (!abort_on_refresh && !result)
   {
-    thd->global_read_lock_protection++;
+    m_protection_count++;
     protect_against_global_read_lock++;
     DBUG_PRINT("sql_lock", ("protect_against_global_read_lock incr: %u",
                             protect_against_global_read_lock));
@@ -1408,7 +1412,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
   @param thd     Reference to thread.
 */
 
-void start_waiting_global_read_lock(THD *thd)
+void Global_read_lock::start_waiting_global_read_lock(THD *thd)
 {
   bool tmp;
   DBUG_ENTER("start_waiting_global_read_lock");
@@ -1416,13 +1420,13 @@ void start_waiting_global_read_lock(THD *thd)
     Ignore request if we do not have protection against global read lock.
     (Note that this is a violation of the interface contract, hence the assert).
   */
-  DBUG_ASSERT(thd->global_read_lock_protection > 0);
-  if (unlikely(thd->global_read_lock_protection == 0))
+  DBUG_ASSERT(m_protection_count > 0);
+  if (unlikely(m_protection_count == 0))
     DBUG_VOID_RETURN;
   /* Decrement local read lock protection counter, return if we still have it */
-  if (unlikely(--thd->global_read_lock_protection > 0))
+  if (unlikely(--m_protection_count > 0))
     DBUG_VOID_RETURN;
-  if (unlikely(thd->global_read_lock))
+  if (unlikely(m_state))
     DBUG_VOID_RETURN;
   (void) pthread_mutex_lock(&LOCK_global_read_lock);
   DBUG_ASSERT(protect_against_global_read_lock);
@@ -1450,7 +1454,7 @@ void start_waiting_global_read_lock(THD *thd)
   @retval True   Failure, thread was killed.
 */
 
-bool make_global_read_lock_block_commit(THD *thd)
+bool Global_read_lock::make_global_read_lock_block_commit(THD *thd)
 {
   bool error;
   const char *old_message;
@@ -1459,7 +1463,7 @@ bool make_global_read_lock_block_commit(THD *thd)
     If we didn't succeed lock_global_read_lock(), or if we already suceeded
     make_global_read_lock_block_commit(), do nothing.
   */
-  if (thd->global_read_lock != GOT_GLOBAL_READ_LOCK)
+  if (m_state != GRL_ACQUIRED)
     DBUG_RETURN(0);
   pthread_mutex_lock(&LOCK_global_read_lock);
   /* increment this BEFORE waiting on cond (otherwise race cond) */
@@ -1476,7 +1480,7 @@ bool make_global_read_lock_block_commit(THD *thd)
   if ((error= test(thd->killed)))
     global_read_lock_blocks_commit--; // undo what we did
   else
-    thd->global_read_lock= MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT;
+    m_state= GRL_ACQUIRED_AND_BLOCKS_COMMIT;
   thd->exit_cond(old_message); // this unlocks LOCK_global_read_lock
   DBUG_RETURN(error);
 }
diff --git a/sql/mdl.cc b/sql/mdl.cc
index dce917a1a2e..3148571131a 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -19,6 +19,10 @@
 #include <hash.h>
 #include <mysqld_error.h>
 
+
+void notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket);
+
+
 static bool mdl_initialized= 0;
 
 
@@ -46,6 +50,41 @@ private:
 };
 
 
+enum enum_deadlock_weight
+{
+  MDL_DEADLOCK_WEIGHT_DML= 0,
+  MDL_DEADLOCK_WEIGHT_DDL= 100
+};
+
+
+
+/**
+  A context of the recursive traversal through all contexts
+  in all sessions in search for deadlock.
+*/
+
+class Deadlock_detection_context
+{
+public:
+  Deadlock_detection_context(MDL_context *start_arg)
+    : start(start_arg),
+      victim(NULL),
+      current_search_depth(0)
+  { }
+  MDL_context *start;
+  MDL_context *victim;
+  uint current_search_depth;
+  static const uint MAX_SEARCH_DEPTH= 1000;
+};
+
+
+/**
+  Get a bit corresponding to enum_mdl_type value in a granted/waiting bitmaps
+  and compatibility matrices.
+*/
+
+#define MDL_BIT(A) static_cast<MDL_lock::bitmap_t>(1U << A)
+
 /**
   The lock context. Created internally for an acquired lock.
   For a given name, there exists only one MDL_lock instance,
@@ -60,53 +99,98 @@ private:
 class MDL_lock
 {
 public:
-  typedef I_P_List<MDL_ticket,
-                   I_P_List_adapter<MDL_ticket,
-                                    &MDL_ticket::next_in_lock,
-                                    &MDL_ticket::prev_in_lock> >
-          Ticket_list;
+  typedef uchar bitmap_t;
 
-  typedef Ticket_list::Iterator Ticket_iterator;
+  class Ticket_list
+  {
+  public:
+    typedef I_P_List<MDL_ticket,
+                     I_P_List_adapter<MDL_ticket,
+                                      &MDL_ticket::next_in_lock,
+                                      &MDL_ticket::prev_in_lock> >
+            List;
+    operator const List &() const { return m_list; }
+    Ticket_list() :m_bitmap(0) {}
+
+    void add_ticket(MDL_ticket *ticket);
+    void remove_ticket(MDL_ticket *ticket);
+    bool is_empty() const { return m_list.is_empty(); }
+    bitmap_t bitmap() const { return m_bitmap; }
+  private:
+    void clear_bit_if_not_in_list(enum_mdl_type type);
+  private:
+    /** List of tickets. */
+    List m_list;
+    /** Bitmap of types of tickets in this list. */
+    bitmap_t m_bitmap;
+  };
+
+  typedef Ticket_list::List::Iterator Ticket_iterator;
 
 public:
   /** The key of the object (data) being protected. */
   MDL_key key;
-  /** List of granted tickets for this lock. */
-  Ticket_list granted;
-  /** Tickets for contexts waiting to acquire a shared lock. */
-  Ticket_list waiting_shared;
-  /**
-    Tickets for contexts waiting to acquire an exclusive lock.
-    There can be several upgraders and active exclusive
-    locks belonging to the same context. E.g.
-    in case of RENAME t1 to t2, t2 to t3, we attempt to
-    exclusively lock t2 twice.
-  */
-  Ticket_list waiting_exclusive;
   void   *cached_object;
   mdl_cached_object_release_hook cached_object_release_hook;
-  /** Mutex protecting this lock context. */
-  pthread_mutex_t m_mutex;
+  /**
+    Read-write lock protecting this lock context.
+
+    TODO/FIXME: Replace with RW-lock which will prefer readers
+                on all platforms and not only on Linux.
+  */
+  rw_lock_t m_rwlock;
 
   bool is_empty() const
   {
-    return (granted.is_empty() && waiting_shared.is_empty() &&
-            waiting_exclusive.is_empty());
+    return (m_granted.is_empty() && m_waiting.is_empty());
   }
 
-  bool has_pending_exclusive_lock()
+  virtual const bitmap_t *incompatible_granted_types_bitmap() const = 0;
+  virtual const bitmap_t *incompatible_waiting_types_bitmap() const = 0;
+
+  bool has_pending_conflicting_lock(enum_mdl_type type);
+
+  bool can_grant_lock(enum_mdl_type type, MDL_context *requstor_ctx) const;
+
+  inline static MDL_lock *create(const MDL_key *key);
+
+  void notify_shared_locks(MDL_context *ctx)
   {
-    bool has_locks;
-    pthread_mutex_lock(&m_mutex);
-    has_locks= ! waiting_exclusive.is_empty();
-    pthread_mutex_unlock(&m_mutex);
-    return has_locks;
+    Ticket_iterator it(m_granted);
+    MDL_ticket *conflicting_ticket;
+
+    while ((conflicting_ticket= it++))
+    {
+      if (conflicting_ticket->get_ctx() != ctx)
+        notify_shared_lock(ctx->get_thd(), conflicting_ticket);
+    }
   }
-  virtual bool can_grant_lock(const MDL_context *requestor_ctx,
-                              enum_mdl_type type, bool is_upgrade)= 0;
-  virtual void wake_up_waiters()= 0;
 
-  inline static MDL_lock *create(const MDL_key *key);
+  /**
+    Wake up contexts which are waiting to acquire lock on the object and
+    which may succeed now, when we released some lock on it or removed
+    some pending request from its waiters list (the latter can happen,
+    for example, when context trying to acquire exclusive on the object
+    lock is killed).
+  */
+  void wake_up_waiters()
+  {
+    MDL_lock::Ticket_iterator it(m_waiting);
+    MDL_ticket *awake_ticket;
+
+    while ((awake_ticket= it++))
+      awake_ticket->get_ctx()->awake(MDL_context::NORMAL_WAKE_UP);
+  }
+  void remove_ticket(Ticket_list MDL_lock::*queue, MDL_ticket *ticket);
+
+  bool find_deadlock(MDL_ticket *waiting_ticket,
+                     Deadlock_detection_context *deadlock_ctx);
+
+  /** List of granted tickets for this lock. */
+  Ticket_list m_granted;
+  /** Tickets for contexts waiting to acquire a lock. */
+  Ticket_list m_waiting;
+public:
 
   MDL_lock(const MDL_key *key_arg)
   : key(key_arg),
@@ -116,31 +200,31 @@ public:
     m_ref_release(0),
     m_is_destroyed(FALSE)
   {
-    pthread_mutex_init(&m_mutex, NULL);
+    my_rwlock_init(&m_rwlock, NULL);
   }
 
   virtual ~MDL_lock()
   {
-    pthread_mutex_destroy(&m_mutex);
+    rwlock_destroy(&m_rwlock);
   }
   inline static void destroy(MDL_lock *lock);
 public:
   /**
     These three members are used to make it possible to separate
-    the mdl_locks.m_mutex mutex and MDL_lock::m_mutex in
+    the mdl_locks.m_mutex mutex and MDL_lock::m_rwlock in
     MDL_map::find_or_insert() for increased scalability.
     The 'm_is_destroyed' member is only set by destroyers that
-    have both the mdl_locks.m_mutex and MDL_lock::m_mutex, thus
+    have both the mdl_locks.m_mutex and MDL_lock::m_rwlock, thus
     holding any of the mutexes is sufficient to read it.
     The 'm_ref_usage; is incremented under protection by
     mdl_locks.m_mutex, but when 'm_is_destroyed' is set to TRUE, this
-    member is moved to be protected by the MDL_lock::m_mutex.
+    member is moved to be protected by the MDL_lock::m_rwlock.
     This means that the MDL_map::find_or_insert() which only
-    holds the MDL_lock::m_mutex can compare it to 'm_ref_release'
+    holds the MDL_lock::m_rwlock can compare it to 'm_ref_release'
     without acquiring mdl_locks.m_mutex again and if equal it can also
     destroy the lock object safely.
     The 'm_ref_release' is incremented under protection by
-    MDL_lock::m_mutex.
+    MDL_lock::m_rwlock.
     Note since we are only interested in equality of these two
     counters we don't have to worry about overflows as long as
     their size is big enough to hold maximum number of concurrent
@@ -164,9 +248,18 @@ public:
     : MDL_lock(key_arg)
   { }
 
-  virtual bool can_grant_lock(const MDL_context *requestor_ctx,
-                              enum_mdl_type type, bool is_upgrade);
-  virtual void wake_up_waiters();
+  virtual const bitmap_t *incompatible_granted_types_bitmap() const
+  {
+    return m_granted_incompatible;
+  }
+  virtual const bitmap_t *incompatible_waiting_types_bitmap() const
+  {
+    return m_waiting_incompatible;
+  }
+
+private:
+  static const bitmap_t m_granted_incompatible[MDL_TYPE_END];
+  static const bitmap_t m_waiting_incompatible[MDL_TYPE_END];
 };
 
 
@@ -182,9 +275,18 @@ public:
     : MDL_lock(key_arg)
   { }
 
-  virtual bool can_grant_lock(const MDL_context *requestor_ctx,
-                              enum_mdl_type type, bool is_upgrade);
-  virtual void wake_up_waiters();
+  virtual const bitmap_t *incompatible_granted_types_bitmap() const
+  {
+    return m_granted_incompatible;
+  }
+  virtual const bitmap_t *incompatible_waiting_types_bitmap() const
+  {
+    return m_waiting_incompatible;
+  }
+
+private:
+  static const bitmap_t m_granted_incompatible[MDL_TYPE_END];
+  static const bitmap_t m_waiting_incompatible[MDL_TYPE_END];
 };
 
 
@@ -275,7 +377,7 @@ void MDL_map::destroy()
   if it does not exist.
 
   @retval non-NULL - Success. MDL_lock instance for the key with
-                     locked MDL_lock::m_mutex.
+                     locked MDL_lock::m_rwlock.
   @retval NULL     - Failure (OOM).
 */
 
@@ -309,7 +411,7 @@ retry:
   Find MDL_lock object corresponding to the key.
 
   @retval non-NULL - MDL_lock instance for the key with locked
-                     MDL_lock::m_mutex.
+                     MDL_lock::m_rwlock.
   @retval NULL     - There was no MDL_lock for the key.
 */
 
@@ -335,7 +437,7 @@ retry:
 
 
 /**
-  Release mdl_locks.m_mutex mutex and lock MDL_lock::m_mutex for lock
+  Release mdl_locks.m_mutex mutex and lock MDL_lock::m_rwlock for lock
   object from the hash. Handle situation when object was released
   while the held no mutex.
 
@@ -357,7 +459,7 @@ bool MDL_map::move_from_hash_to_lock_mutex(MDL_lock *lock)
   lock->m_ref_usage++;
   pthread_mutex_unlock(&m_mutex);
 
-  pthread_mutex_lock(&lock->m_mutex);
+  rw_wrlock(&lock->m_rwlock);
   lock->m_ref_release++;
   if (unlikely(lock->m_is_destroyed))
   {
@@ -372,7 +474,7 @@ bool MDL_map::move_from_hash_to_lock_mutex(MDL_lock *lock)
     */
     uint ref_usage= lock->m_ref_usage;
     uint ref_release= lock->m_ref_release;
-    pthread_mutex_unlock(&lock->m_mutex);
+    rw_unlock(&lock->m_rwlock);
     if (ref_usage == ref_release)
       MDL_lock::destroy(lock);
     return TRUE;
@@ -391,8 +493,6 @@ void MDL_map::remove(MDL_lock *lock)
 {
   uint ref_usage, ref_release;
 
-  safe_mutex_assert_owner(&lock->m_mutex);
-
   if (lock->cached_object)
     (*lock->cached_object_release_hook)(lock->cached_object);
 
@@ -402,14 +502,14 @@ void MDL_map::remove(MDL_lock *lock)
     has the responsibility to release it.
 
     Setting of m_is_destroyed to TRUE while holding _both_
-    mdl_locks.m_mutex and MDL_lock::m_mutex mutexes transfers the
+    mdl_locks.m_mutex and MDL_lock::m_rwlock mutexes transfers the
     protection of m_ref_usage from mdl_locks.m_mutex to
-    MDL_lock::m_mutex while removal of object from the hash makes
-    it read-only.  Therefore whoever acquires MDL_lock::m_mutex next
+    MDL_lock::m_rwlock while removal of object from the hash makes
+    it read-only.  Therefore whoever acquires MDL_lock::m_rwlock next
     will see most up to date version of m_ref_usage.
 
     This means that when m_is_destroyed is TRUE and we hold the
-    MDL_lock::m_mutex we can safely read the m_ref_usage
+    MDL_lock::m_rwlock we can safely read the m_ref_usage
     member.
   */
   pthread_mutex_lock(&m_mutex);
@@ -417,7 +517,7 @@ void MDL_map::remove(MDL_lock *lock)
   lock->m_is_destroyed= TRUE;
   ref_usage= lock->m_ref_usage;
   ref_release= lock->m_ref_release;
-  pthread_mutex_unlock(&lock->m_mutex);
+  rw_unlock(&lock->m_rwlock);
   pthread_mutex_unlock(&m_mutex);
   if (ref_usage == ref_release)
     MDL_lock::destroy(lock);
@@ -431,10 +531,16 @@ void MDL_map::remove(MDL_lock *lock)
 */
 
 MDL_context::MDL_context()
-  :m_lt_or_ha_sentinel(NULL),
-  m_thd(NULL)
+  :m_trans_sentinel(NULL),
+  m_thd(NULL),
+  m_needs_thr_lock_abort(FALSE),
+  m_waiting_for(NULL),
+  m_deadlock_weight(0),
+  m_signal(NO_WAKE_UP)
 {
-  pthread_cond_init(&m_ctx_wakeup_cond, NULL);
+  my_rwlock_init(&m_waiting_for_lock, NULL);
+  pthread_mutex_init(&m_signal_lock, NULL);
+  pthread_cond_init(&m_signal_cond, NULL);
 }
 
 
@@ -453,7 +559,10 @@ MDL_context::MDL_context()
 void MDL_context::destroy()
 {
   DBUG_ASSERT(m_tickets.is_empty());
-  pthread_cond_destroy(&m_ctx_wakeup_cond);
+
+  rwlock_destroy(&m_waiting_for_lock);
+  pthread_mutex_destroy(&m_signal_lock);
+  pthread_cond_destroy(&m_signal_cond);
 }
 
 
@@ -540,6 +649,13 @@ MDL_request::create(MDL_key::enum_mdl_namespace mdl_namespace, const char *db,
 }
 
 
+uint MDL_request::get_deadlock_weight() const
+{
+  return key.mdl_namespace() == MDL_key::GLOBAL ||
+         type > MDL_SHARED_NO_WRITE ?
+    MDL_DEADLOCK_WEIGHT_DDL : MDL_DEADLOCK_WEIGHT_DML;
+}
+
 /**
   Auxiliary functions needed for creation/destruction of MDL_lock objects.
 
@@ -567,8 +683,6 @@ void MDL_lock::destroy(MDL_lock *lock)
 }
 
 
-
-
 /**
   Auxiliary functions needed for creation/destruction of MDL_ticket
   objects.
@@ -650,485 +764,463 @@ static inline void mdl_exit_cond(THD *thd,
 }
 
 
-/**
-  Check if request for the global metadata lock can be satisfied given
-  its current state,
-
-  @param  requestor_ctx  The context that identifies the owner of the request.
-  @param  type_arg       The requested type of global lock. Usually derived
-                         from the type of lock on individual object to be
-                         requested. See table below.
-  @param  is_upgrade     TRUE if we are performing lock upgrade (not unused).
-
-  @retval TRUE  - Lock request can be satisfied
-  @retval FALSE - There is some conflicting lock
-
-  Here is a compatibility matrix defined by this function:
-
-                  |             | Satisfied or pending requests
-                  |             | for global metadata lock
-  ----------------+-------------+--------------------------------------------
-  Type of request | Correspond. |
-  for indiv. lock | global lock | Active-S  Pending-S  Active-IS(**) Active-IX
-  ----------------+-------------+--------------------------------------------
-  S, high-prio S  |   IS        |    +         +          +             +
-  upgradable S    |   IX        |    -         -          +             +
-  X               |   IX        |    -         -          +             +
-  S upgraded to X |   IX (*)    |    0         +          +             +
+MDL_context::mdl_signal_type MDL_context::wait()
+{
+  const char *old_msg;
+  st_my_thread_var *mysys_var= my_thread_var;
+  mdl_signal_type result;
 
-  Here: "+" -- means that request can be satisfied
-        "-" -- means that request can't be satisfied and should wait
-        "0" -- means impossible situation.
+  pthread_mutex_lock(&m_signal_lock);
 
-  (*)  Since for upgradable shared locks we always take intention exclusive
-       global lock at the same time when obtaining the shared lock, there
-       is no need to obtain such lock during the upgrade itself.
-  (**) Since intention shared global locks are compatible with all other
-       type of locks we don't even have any accounting for them.
-*/
+  old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_signal_cond, &m_signal_lock);
 
-bool
-MDL_global_lock::can_grant_lock(const MDL_context *requestor_ctx,
-                                enum_mdl_type type_arg,
-                                bool is_upgrade)
+  while (! m_signal && !mysys_var->abort)
+    pthread_cond_wait(&m_signal_cond, &m_signal_lock);
+
+  result= m_signal;
+
+  MDL_EXIT_COND(m_thd, mysys_var, &m_signal_lock, old_msg);
+
+  return result;
+}
+
+
+MDL_context::mdl_signal_type MDL_context::timed_wait(ulong timeout)
 {
-  switch (type_arg)
+  struct timespec abstime;
+  const char *old_msg;
+  mdl_signal_type result;
+  st_my_thread_var *mysys_var= my_thread_var;
+
+  pthread_mutex_lock(&m_signal_lock);
+
+  old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_signal_cond, &m_signal_lock);
+
+  if (! m_signal)
   {
-  case MDL_SHARED:
-    if (! granted.is_empty() && granted.front()->m_type == MDL_INTENTION_EXCLUSIVE)
-    {
-      /*
-        We are going to obtain global shared lock and there is active
-        intention exclusive lock. Have to wait.
-      */
-      return FALSE;
-    }
-    return TRUE;
-    break;
-  case MDL_INTENTION_EXCLUSIVE:
-    if ((! granted.is_empty() && granted.front()->m_type == MDL_SHARED) ||
-        ! waiting_shared.is_empty())
-    {
-      /*
-        We are going to obtain intention exclusive global lock and
-        there is active or pending shared global lock. Have to wait.
-      */
-      return FALSE;
-    }
-    else
-      return TRUE;
-    break;
-  default:
-    DBUG_ASSERT(0);
-    break;
+    set_timespec(abstime, timeout);
+    pthread_cond_timedwait(&m_signal_cond, &m_signal_lock, &abstime);
   }
-  return FALSE;
+
+  result= (m_signal != NO_WAKE_UP) ? m_signal : TIMEOUT_WAKE_UP;
+
+  MDL_EXIT_COND(m_thd, mysys_var, &m_signal_lock, old_msg);
+
+  return result;
 }
 
 
 /**
-  Wake up contexts which are waiting to acquire the global
-  metadata lock and which may succeed now, when we released it, or
-  removed a blocking request for it from the waiters list.
-  The latter can happen when the context trying to acquire the
-  global shared lock is killed.
+  Clear bit corresponding to the type of metadata lock in bitmap representing
+  set of such types if list of tickets does not contain ticket with such type.
+
+  @param[in,out]  bitmap  Bitmap representing set of types of locks.
+  @param[in]      list    List to inspect.
+  @param[in]      type    Type of metadata lock to look up in the list.
 */
 
-void MDL_global_lock::wake_up_waiters()
+void MDL_lock::Ticket_list::clear_bit_if_not_in_list(enum_mdl_type type)
 {
-  /*
-    If there are no active locks or they are of INTENTION
-    EXCLUSIVE type and there are no pending requests for global
-    SHARED lock, wake up contexts waiting for an INTENTION
-    EXCLUSIVE lock.
-    This happens when we release the global SHARED lock or abort
-    or remove a pending request for it, i.e. abort the
-    context waiting for it.
-  */
-  if ((granted.is_empty() ||
-       granted.front()->m_type == MDL_INTENTION_EXCLUSIVE) &&
-      waiting_shared.is_empty() && ! waiting_exclusive.is_empty())
-  {
-    MDL_lock::Ticket_iterator it(waiting_exclusive);
-    MDL_ticket *awake_ticket;
-    while ((awake_ticket= it++))
-      awake_ticket->get_ctx()->awake();
-  }
+  MDL_lock::Ticket_iterator it(m_list);
+  const MDL_ticket *ticket;
 
+  while ((ticket= it++))
+    if (ticket->get_type() == type)
+      return;
+  m_bitmap&= ~ MDL_BIT(type);
+}
+
+
+/**
+  Add ticket to MDL_lock's list of waiting requests and
+  update corresponding bitmap of lock types.
+*/
+
+void MDL_lock::Ticket_list::add_ticket(MDL_ticket *ticket)
+{
+  m_list.push_front(ticket);
+  m_bitmap|= MDL_BIT(ticket->get_type());
+}
+
+
+/**
+  Remove ticket from MDL_lock's list of requests and
+  update corresponding bitmap of lock types.
+*/
+
+void MDL_lock::Ticket_list::remove_ticket(MDL_ticket *ticket)
+{
+  m_list.remove(ticket);
   /*
-    If there are no active locks, wake up contexts waiting for
-    the global shared lock (happens when an INTENTION EXCLUSIVE
-    lock is released).
-
-    We don't wake up contexts waiting for the global shared lock
-    if there is an active global shared lock since such situation
-    is transient and in it contexts marked as waiting for global
-    shared lock must be already woken up and simply have not
-    managed to update lock object yet.
+    Check if waiting queue has another ticket with the same type as
+    one which was removed. If there is no such ticket, i.e. we have
+    removed last ticket of particular type, then we need to update
+    bitmap of waiting ticket's types.
+    Note that in most common case, i.e. when shared lock is removed
+    from waiting queue, we are likely to find ticket of the same
+    type early without performing full iteration through the list.
+    So this method should not be too expensive.
   */
-  if (granted.is_empty() &&
-      ! waiting_shared.is_empty())
-  {
-    MDL_lock::Ticket_iterator it(waiting_shared);
-    MDL_ticket *awake_ticket;
-    while ((awake_ticket= it++))
-      awake_ticket->get_ctx()->awake();
-  }
+  clear_bit_if_not_in_list(ticket->get_type());
 }
 
 
 /**
-  Check if request for the per-object lock can be satisfied given current
-  state of the lock.
+  Compatibility (or rather "incompatibility") matrices for global metadata
+  lock. Arrays of bitmaps which elements specify which granted/waiting locks
+  are incompatible with type of lock being requested.
+
+  Here is how types of individual locks are translated to type of global lock:
+
+    ----------------+-------------+
+    Type of request | Correspond. |
+    for indiv. lock | global lock |
+    ----------------+-------------+
+    S, SH, SR, SW   |   IS        |
+    SNW, SNRW, X    |   IX        |
+    SNW, SNRW -> X  |   IX (*)    |
+
+  The first array specifies if particular type of request can be satisfied
+  if there is granted global lock of certain type.
+
+             | Type of active |
+     Request |   global lock  |
+      type   | IS(**) IX   S  |
+    ---------+----------------+
+    IS       |  +      +   +  |
+    IX       |  +      +   -  |
+    S        |  +      -   +  |
+
+  The second array specifies if particular type of request can be satisfied
+  if there is already waiting request for the global lock of certain type.
+  I.e. it specifies what is the priority of different lock types.
+
+             |    Pending   |
+     Request |  global lock |
+      type   | IS(**) IX  S |
+    ---------+--------------+
+    IS       |  +      +  + |
+    IX       |  +      +  - |
+    S        |  +      +  + |
 
-  @param  requestor_ctx  The context that identifies the owner of the request.
-  @param  type_arg       The requested lock type.
-  @param  is_upgrade     Must be set to TRUE when we are upgrading
-                         a shared upgradable lock to exclusive.
+  Here: "+" -- means that request can be satisfied
+        "-" -- means that request can't be satisfied and should wait
 
-  @retval TRUE   Lock request can be satisfied
-  @retval FALSE  There is some conflicting lock.
+  (*)   Since for upgradable locks we always take intention exclusive global
+        lock at the same time when obtaining the shared lock, there is no
+        need to obtain such lock during the upgrade itself.
+  (**)  Since intention shared global locks are compatible with all other
+        type of locks we don't even have any accounting for them.
+*/
+
+const MDL_lock::bitmap_t MDL_global_lock::m_granted_incompatible[MDL_TYPE_END] =
+{
+  MDL_BIT(MDL_SHARED), MDL_BIT(MDL_INTENTION_EXCLUSIVE), 0, 0, 0, 0, 0, 0
+};
 
-  This function defines the following compatibility matrix for metadata locks:
+const MDL_lock::bitmap_t MDL_global_lock::m_waiting_incompatible[MDL_TYPE_END] =
+{
+  MDL_BIT(MDL_SHARED), 0, 0, 0, 0, 0, 0, 0
+};
 
-                  | Satisfied or pending requests which we have in MDL_lock
-  ----------------+---------------------------------------------------------
-  Current request | Active-S  Pending-X Active-X Act-S-pend-upgrade-to-X
-  ----------------+---------------------------------------------------------
-  S, upgradable S |    +         -         - (*)           -
-  High-prio S     |    +         +         -               +
-  X               |    -         +         -               -
-  S upgraded to X |    - (**)    +         0               0
+
+/**
+  Compatibility (or rather "incompatibility") matrices for per-object
+  metadata lock. Arrays of bitmaps which elements specify which granted/
+  waiting locks are incompatible with type of lock being requested.
+
+  The first array specifies if particular type of request can be satisfied
+  if there is granted lock of certain type.
+
+     Request  |  Granted requests for lock   |
+      type    | S  SH  SR  SW  SNW  SNRW  X  |
+    ----------+------------------------------+
+    S         | +   +   +   +   +    +    -  |
+    SH        | +   +   +   +   +    +    -  |
+    SR        | +   +   +   +   +    -    -  |
+    SW        | +   +   +   +   -    -    -  |
+    SNW       | +   +   +   -   -    -    -  |
+    SNRW      | +   +   -   -   -    -    -  |
+    X         | -   -   -   -   -    -    -  |
+    SNW -> X  | -   -   -   0   0    0    0  |
+    SNRW -> X | -   -   0   0   0    0    0  |
+
+  The second array specifies if particular type of request can be satisfied
+  if there is waiting request for the same lock of certain type. In other
+  words it specifies what is the priority of different lock types.
+
+     Request  |  Pending requests for lock  |
+      type    | S  SH  SR  SW  SNW  SNRW  X |
+    ----------+-----------------------------+
+    S         | +   +   +   +   +     +   - |
+    SH        | +   +   +   +   +     +   + |
+    SR        | +   +   +   +   +     -   - |
+    SW        | +   +   +   +   -     -   - |
+    SNW       | +   +   +   +   +     +   - |
+    SNRW      | +   +   +   +   +     +   - |
+    X         | +   +   +   +   +     +   + |
+    SNW -> X  | +   +   +   +   +     +   + |
+    SNRW -> X | +   +   +   +   +     +   + |
 
   Here: "+" -- means that request can be satisfied
         "-" -- means that request can't be satisfied and should wait
         "0" -- means impossible situation which will trigger assert
 
-  (*)  Unless active exclusive lock belongs to the same context as shared
-       lock being requested.
-  (**) Unless all active shared locks belong to the same context as one
-       being upgraded.
+  @note In cases then current context already has "stronger" type
+        of lock on the object it will be automatically granted
+        thanks to usage of the MDL_context::find_ticket() method.
+*/
+
+const MDL_lock::bitmap_t
+MDL_object_lock::m_granted_incompatible[MDL_TYPE_END] =
+{
+  0,
+  MDL_BIT(MDL_EXCLUSIVE),
+  MDL_BIT(MDL_EXCLUSIVE),
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED_NO_READ_WRITE),
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED_NO_READ_WRITE) |
+    MDL_BIT(MDL_SHARED_NO_WRITE),
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED_NO_READ_WRITE) |
+    MDL_BIT(MDL_SHARED_NO_WRITE) | MDL_BIT(MDL_SHARED_WRITE),
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED_NO_READ_WRITE) |
+    MDL_BIT(MDL_SHARED_NO_WRITE) | MDL_BIT(MDL_SHARED_WRITE) |
+    MDL_BIT(MDL_SHARED_READ),
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED_NO_READ_WRITE) |
+    MDL_BIT(MDL_SHARED_NO_WRITE) | MDL_BIT(MDL_SHARED_WRITE) |
+    MDL_BIT(MDL_SHARED_READ) | MDL_BIT(MDL_SHARED_HIGH_PRIO) |
+    MDL_BIT(MDL_SHARED)
+};
+
+
+const MDL_lock::bitmap_t
+MDL_object_lock::m_waiting_incompatible[MDL_TYPE_END] =
+{
+  0,
+  MDL_BIT(MDL_EXCLUSIVE),
+  0,
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED_NO_READ_WRITE),
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED_NO_READ_WRITE) |
+    MDL_BIT(MDL_SHARED_NO_WRITE),
+  MDL_BIT(MDL_EXCLUSIVE),
+  MDL_BIT(MDL_EXCLUSIVE),
+  0
+};
+
+
+/**
+  Check if request for the metadata lock can be satisfied given its
+  current state.
+
+  @param  type_arg       The requested lock type.
+  @param  requestor_ctx  The MDL context of the requestor.
+
+  @retval TRUE   Lock request can be satisfied
+  @retval FALSE  There is some conflicting lock.
+
+  @note In cases then current context already has "stronger" type
+        of lock on the object it will be automatically granted
+        thanks to usage of the MDL_context::find_ticket() method.
 */
 
 bool
-MDL_object_lock::can_grant_lock(const MDL_context *requestor_ctx,
-                                enum_mdl_type type_arg,
-                                bool is_upgrade)
+MDL_lock::can_grant_lock(enum_mdl_type type_arg,
+                         MDL_context *requestor_ctx) const
 {
   bool can_grant= FALSE;
-
-  switch (type_arg) {
-  case MDL_SHARED:
-  case MDL_SHARED_UPGRADABLE:
-  case MDL_SHARED_HIGH_PRIO:
-    if (granted.is_empty() || granted.front()->is_shared())
-    {
-      /* Pending exclusive locks have higher priority over shared locks. */
-      if (waiting_exclusive.is_empty() || type_arg == MDL_SHARED_HIGH_PRIO)
-        can_grant= TRUE;
-    }
-    else if (granted.front()->get_ctx() == requestor_ctx)
-    {
-      /*
-        When exclusive lock comes from the same context we can satisfy our
-        shared lock. This is required for CREATE TABLE ... SELECT ... and
-        ALTER VIEW ... AS ....
-      */
+  bitmap_t waiting_incompat_map= incompatible_waiting_types_bitmap()[type_arg];
+  bitmap_t granted_incompat_map= incompatible_granted_types_bitmap()[type_arg];
+  /*
+    New lock request can be satisfied iff:
+    - There are no incompatible types of satisfied requests
+    in other contexts
+    - There are no waiting requests which have higher priority
+    than this request.
+  */
+  if (! (m_waiting.bitmap() & waiting_incompat_map))
+  {
+    if (! (m_granted.bitmap() & granted_incompat_map))
       can_grant= TRUE;
-    }
-    break;
-  case MDL_EXCLUSIVE:
-    if (is_upgrade)
+    else
     {
-      /* We are upgrading MDL_SHARED to MDL_EXCLUSIVE. */
-      MDL_ticket *conflicting_ticket;
-      MDL_lock::Ticket_iterator it(granted);
+      Ticket_iterator it(m_granted);
+      MDL_ticket *ticket;
 
-      /*
-        There should be no active exclusive locks since we own shared lock
-        on the object.
-      */
-      DBUG_ASSERT(granted.front()->is_shared());
-
-      while ((conflicting_ticket= it++))
+      /* Check that the incompatible lock belongs to some other context. */
+      while ((ticket= it++))
       {
-        /*
-          When upgrading shared lock to exclusive one we can have other shared
-          locks for the same object in the same context, e.g. in case when several
-          instances of TABLE are open.
-        */
-        if (conflicting_ticket->get_ctx() != requestor_ctx)
+        if (ticket->get_ctx() != requestor_ctx &&
+            ticket->is_incompatible_when_granted(type_arg))
           break;
       }
-      /* Grant lock if there are no conflicting shared locks. */
-      if (conflicting_ticket == NULL)
+      if (ticket == NULL)             /* Incompatible locks are our own. */
         can_grant= TRUE;
-      break;
-    }
-    else if (granted.is_empty())
-    {
-      /*
-        We are trying to acquire fresh MDL_EXCLUSIVE and there are no active
-        shared or exclusive locks.
-      */
-      can_grant= TRUE;
     }
-    break;
-  default:
-    DBUG_ASSERT(0);
   }
   return can_grant;
 }
 
 
-/**
-  Wake up contexts which are waiting to acquire lock on individual object
-  and which may succeed now, when we released some lock on it or removed
-  some pending request from its waiters list (the latter can happen, for
-  example, when context trying to acquire exclusive lock is killed).
-*/
+/** Remove a ticket from waiting or pending queue and wakeup up waiters. */
 
-void MDL_object_lock::wake_up_waiters()
+void MDL_lock::remove_ticket(Ticket_list MDL_lock::*list, MDL_ticket *ticket)
 {
-  /*
-    There are no active locks or they are of shared type.
-    We have to wake up contexts waiting for shared lock even if there is
-    a pending exclusive lock as some them might be trying to acquire high
-    priority shared lock.
-  */
-  if ((granted.is_empty() || granted.front()->is_shared()) &&
-      ! waiting_shared.is_empty())
-  {
-    MDL_lock::Ticket_iterator it(waiting_shared);
-    MDL_ticket *waiting_ticket;
-    while ((waiting_ticket= it++))
-      waiting_ticket->get_ctx()->awake();
-  }
-
-  /*
-    There are no active locks (shared or exclusive).
-    Wake up contexts waiting to acquire exclusive locks.
-  */
-  if (granted.is_empty() && ! waiting_exclusive.is_empty())
+  rw_wrlock(&m_rwlock);
+  (this->*list).remove_ticket(ticket);
+  if (is_empty())
+    mdl_locks.remove(this);
+  else
   {
-    MDL_lock::Ticket_iterator it(waiting_exclusive);
-    MDL_ticket *waiting_ticket;
-    while ((waiting_ticket= it++))
-      waiting_ticket->get_ctx()->awake();
+    /*
+      There can be some contexts waiting to acquire a lock
+      which now might be able to do it. Wake them up!
+    */
+    wake_up_waiters();
+    rw_unlock(&m_rwlock);
   }
 }
 
 
 /**
-  Check whether the context already holds a compatible lock ticket
-  on an object.
-  Start searching the transactional locks. If not
-  found in the list of transactional locks, look at LOCK TABLES
-  and HANDLER locks.
+  Check if we have any pending locks which conflict with existing
+  shared lock.
 
-  @param mdl_request  Lock request object for lock to be acquired
-  @param[out] is_lt_or_ha  Did we pass beyond m_lt_or_ha_sentinel while
-                            searching for ticket?
+  @pre The ticket must match an acquired lock.
 
-  @return A pointer to the lock ticket for the object or NULL otherwise.
+  @return TRUE if there is a conflicting lock request, FALSE otherwise.
 */
 
-MDL_ticket *
-MDL_context::find_ticket(MDL_request *mdl_request,
-                         bool *is_lt_or_ha)
+bool MDL_lock::has_pending_conflicting_lock(enum_mdl_type type)
 {
-  MDL_ticket *ticket;
-  Ticket_iterator it(m_tickets);
-
-  *is_lt_or_ha= FALSE;
-
-  while ((ticket= it++))
-  {
-    if (ticket == m_lt_or_ha_sentinel)
-      *is_lt_or_ha= TRUE;
+  bool result;
 
-    if (mdl_request->type == ticket->m_type &&
-        mdl_request->key.is_equal(&ticket->m_lock->key))
-      break;
-  }
+  safe_mutex_assert_not_owner(&LOCK_open);
 
-  return ticket;
+  rw_rdlock(&m_rwlock);
+  result= (m_waiting.bitmap() & incompatible_granted_types_bitmap()[type]);
+  rw_unlock(&m_rwlock);
+  return result;
 }
 
 
 /**
-  Try to acquire global intention exclusive lock.
+  Check if ticket represents metadata lock of "stronger" or equal type
+  than specified one. I.e. if metadata lock represented by ticket won't
+  allow any of locks which are not allowed by specified type of lock.
 
-  @param[in/out]  mdl_request  Lock request object for lock to be acquired
-
-  @retval  FALSE   Success. The lock may have not been acquired.
-                   One needs to check value of 'MDL_request::ticket'
-                   to find out what has happened.
-  @retval  TRUE    Error.
+  @return TRUE  if ticket has stronger or equal type
+          FALSE otherwise.
 */
 
-bool
-MDL_context::
-try_acquire_global_intention_exclusive_lock(MDL_request *mdl_request)
+bool MDL_ticket::has_stronger_or_equal_type(enum_mdl_type type) const
 {
-  DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::GLOBAL &&
-              mdl_request->type == MDL_INTENTION_EXCLUSIVE);
+  const MDL_lock::bitmap_t *
+    granted_incompat_map= m_lock->incompatible_granted_types_bitmap();
+
+  return ! (granted_incompat_map[type] & ~(granted_incompat_map[m_type]));
+}
 
-  if (is_global_lock_owner(MDL_SHARED))
-  {
-    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
-    return TRUE;
-  }
 
-  return try_acquire_lock_impl(mdl_request);
+bool MDL_ticket::is_incompatible_when_granted(enum_mdl_type type) const
+{
+  return (MDL_BIT(m_type) &
+          m_lock->incompatible_granted_types_bitmap()[type]);
 }
 
 
-/**
-  Acquire one lock with waiting for conflicting locks to go away if needed.
+bool MDL_ticket::is_incompatible_when_waiting(enum_mdl_type type) const
+{
+  return (MDL_BIT(m_type) &
+          m_lock->incompatible_waiting_types_bitmap()[type]);
+}
 
-  @note This is an internal method which should not be used outside of MDL
-        subsystem as in most cases simply waiting for conflicting locks to
-        go away will lead to deadlock.
 
-  @param mdl_request [in/out] Lock request object for lock to be acquired
+/**
+  Acquire global intention exclusive lock.
 
-  @retval  FALSE   Success. MDL_request::ticket points to the ticket
-                   for the lock.
-  @retval  TRUE    Failure (Out of resources or waiting is aborted),
+  @param[in]  mdl_request  Lock request object for lock to be acquired
+
+  @retval  FALSE   Success. The lock has been acquired.
+  @retval  TRUE    Error.
 */
 
 bool
-MDL_context::acquire_lock_impl(MDL_request *mdl_request)
+MDL_context::acquire_global_intention_exclusive_lock(MDL_request *mdl_request)
 {
-  bool not_used;
-  MDL_ticket *ticket;
-  MDL_key *key= &mdl_request->key;
-  MDL_lock *lock;
-  const char *old_msg;
-  st_my_thread_var *mysys_var= my_thread_var;
-
-  DBUG_ASSERT(mdl_request->ticket == NULL);
-  safe_mutex_assert_not_owner(&LOCK_open);
+  DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::GLOBAL &&
+              mdl_request->type == MDL_INTENTION_EXCLUSIVE);
 
   /*
-    Grant lock without waiting if this context already owns this type of lock
-    on this object.
-
-    The fact that we don't wait in such situation allows to avoid deadlocks
-    in cases when pending request for global shared lock pops up after the
-    moment when thread has acquired its first intention exclusive lock but
-    before it has requested the second instance of such lock.
+    If this is a non-recursive attempt to acquire global intention
+    exclusive lock we might have to wait until active global shared
+    lock or pending requests will go away. Since we won't hold any
+    resources (except associated with open HANDLERs) while doing it
+    deadlocks are not possible.
   */
-  if ((mdl_request->ticket= find_ticket(mdl_request, &not_used)))
-    return FALSE;
-
-  if (! (ticket= MDL_ticket::create(this, mdl_request->type)))
-    return TRUE;
-
-  /* The below call also implicitly locks MDL_lock::m_mutex. */
-  if (! (lock= mdl_locks.find_or_insert(key)))
-  {
-    MDL_ticket::destroy(ticket);
-    return TRUE;
-  }
+  DBUG_ASSERT(is_lock_owner(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE) ||
+              ! has_locks() ||
+              (m_trans_sentinel && m_tickets.front() == m_trans_sentinel));
 
-  old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_ctx_wakeup_cond,
-                          &lock->m_mutex);
+  return acquire_lock(mdl_request);
+}
 
-  if (! lock->can_grant_lock(this, mdl_request->type, FALSE))
-  {
-    if (mdl_request->is_shared())
-      lock->waiting_shared.push_front(ticket);
-    else
-      lock->waiting_exclusive.push_front(ticket);
 
-    do
-    {
-      pthread_cond_wait(&m_ctx_wakeup_cond, &lock->m_mutex);
-    }
-    while (! lock->can_grant_lock(this, mdl_request->type, FALSE) &&
-           ! mysys_var->abort);
+/**
+  Check whether the context already holds a compatible lock ticket
+  on an object.
+  Start searching the transactional locks. If not
+  found in the list of transactional locks, look at LOCK TABLES
+  and HANDLER locks.
 
-    if (mysys_var->abort)
-    {
-      /*
-        We have to do MDL_EXIT_COND here and then re-acquire the lock
-        as there is a chance that we will destroy MDL_lock object and
-        won't be able to call MDL_EXIT_COND after it.
-      */
-      MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+  @param mdl_request  Lock request object for lock to be acquired
+  @param[out] is_transactional FALSE if we pass beyond m_trans_sentinel
+                      while searching for ticket, otherwise TRUE.
 
-      pthread_mutex_lock(&lock->m_mutex);
-      /* Get rid of pending ticket. */
-      if (mdl_request->is_shared())
-        lock->waiting_shared.remove(ticket);
-      else
-        lock->waiting_exclusive.remove(ticket);
-      if (lock->is_empty())
-        mdl_locks.remove(lock);
-      else
-      {
-        lock->wake_up_waiters();
-        pthread_mutex_unlock(&lock->m_mutex);
-      }
-      MDL_ticket::destroy(ticket);
-      return TRUE;
-    }
+  @note Tickets which correspond to lock types "stronger" than one
+        being requested are also considered compatible.
 
-    if (mdl_request->is_shared())
-      lock->waiting_shared.remove(ticket);
-    else
-      lock->waiting_exclusive.remove(ticket);
-  }
+  @return A pointer to the lock ticket for the object or NULL otherwise.
+*/
 
-  lock->granted.push_front(ticket);
-  MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+MDL_ticket *
+MDL_context::find_ticket(MDL_request *mdl_request,
+                         bool *is_transactional)
+{
+  MDL_ticket *ticket;
+  Ticket_iterator it(m_tickets);
 
-  ticket->m_state= MDL_ACQUIRED;
-  ticket->m_lock= lock;
+  *is_transactional= TRUE;
 
-  m_tickets.push_front(ticket);
+  while ((ticket= it++))
+  {
+    if (ticket == m_trans_sentinel)
+      *is_transactional= FALSE;
 
-  mdl_request->ticket= ticket;
+    if (mdl_request->key.is_equal(&ticket->m_lock->key) &&
+        ticket->has_stronger_or_equal_type(mdl_request->type))
+      break;
+  }
 
-  return FALSE;
+  return ticket;
 }
 
 
 /**
-  Acquire global intention exclusive lock.
+  Acquire one lock with waiting for conflicting locks to go away if needed.
 
-  @param[in]  mdl_request  Lock request object for lock to be acquired
+  @note This is an internal method which should not be used outside of MDL
+        subsystem as in most cases simply waiting for conflicting locks to
+        go away will lead to deadlock.
 
-  @retval  FALSE   Success. The lock has been acquired.
-  @retval  TRUE    Error.
+  @param mdl_request [in/out] Lock request object for lock to be acquired
+
+  @retval  FALSE   Success. MDL_request::ticket points to the ticket
+                   for the lock.
+  @retval  TRUE    Failure (Out of resources or waiting is aborted),
 */
 
 bool
-MDL_context::acquire_global_intention_exclusive_lock(MDL_request *mdl_request)
+MDL_context::acquire_lock(MDL_request *mdl_request)
 {
-  DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::GLOBAL &&
-              mdl_request->type == MDL_INTENTION_EXCLUSIVE);
-
-  if (is_global_lock_owner(MDL_SHARED))
-  {
-    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
-    return TRUE;
-  }
-
-  /*
-    If this is a non-recursive attempt to acquire global intention
-    exclusive lock we might have to wait until active global shared
-    lock or pending requests will go away. Since we won't hold any
-    resources (except associated with open HANDLERs) while doing it
-    deadlocks are not possible,
-  */
-  DBUG_ASSERT(is_global_lock_owner(MDL_INTENTION_EXCLUSIVE) ||
-              ! has_locks() ||
-              (m_lt_or_ha_sentinel &&
-               m_tickets.front() == m_lt_or_ha_sentinel));
-
   return acquire_lock_impl(mdl_request);
 }
 
@@ -1136,22 +1228,42 @@ MDL_context::acquire_global_intention_exclusive_lock(MDL_request *mdl_request)
 /**
   Try to acquire one lock.
 
+  Unlike exclusive locks, shared locks are acquired one by
+  one. This is interface is chosen to simplify introduction of
+  the new locking API to the system. MDL_context::try_acquire_lock()
+  is currently used from open_table(), and there we have only one
+  table to work with.
+
+  This function may also be used to try to acquire an exclusive
+  lock on a destination table, by ALTER TABLE ... RENAME.
+
+  Returns immediately without any side effect if encounters a lock
+  conflict. Otherwise takes the lock.
+
+  FIXME: Compared to lock_table_name_if_not_cached() (from 5.1)
+         it gives slightly more false negatives.
+
   @param mdl_request [in/out] Lock request object for lock to be acquired
 
   @retval  FALSE   Success. The lock may have not been acquired.
                    Check the ticket, if it's NULL, a conflicting lock
-                   exists.
+                   exists and another attempt should be made after releasing
+                   all current locks and waiting for conflicting lock go
+                   away (using MDL_context::wait_for_lock()).
   @retval  TRUE    Out of resources, an error has been reported.
 */
 
 bool
-MDL_context::try_acquire_lock_impl(MDL_request *mdl_request)
+MDL_context::try_acquire_lock(MDL_request *mdl_request)
 {
   MDL_lock *lock;
   MDL_key *key= &mdl_request->key;
   MDL_ticket *ticket;
-  bool is_lt_or_ha;
+  bool is_transactional;
 
+  DBUG_ASSERT(mdl_request->type < MDL_SHARED_NO_WRITE ||
+              (is_lock_owner(MDL_key::GLOBAL, "", "",
+                             MDL_INTENTION_EXCLUSIVE)));
   DBUG_ASSERT(mdl_request->ticket == NULL);
 
   /* Don't take chances in production. */
@@ -1162,10 +1274,10 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request)
     Check whether the context already holds a shared lock on the object,
     and if so, grant the request.
   */
-  if ((ticket= find_ticket(mdl_request, &is_lt_or_ha)))
+  if ((ticket= find_ticket(mdl_request, &is_transactional)))
   {
-    DBUG_ASSERT(ticket->m_state == MDL_ACQUIRED);
-    DBUG_ASSERT(ticket->m_type == mdl_request->type);
+    DBUG_ASSERT(ticket->m_lock);
+    DBUG_ASSERT(ticket->m_type >= mdl_request->type);
     /*
       If the request is for a transactional lock, and we found
       a transactional lock, just reuse the found ticket.
@@ -1184,7 +1296,7 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request)
       a different alias.
     */
     mdl_request->ticket= ticket;
-    if (is_lt_or_ha && clone_ticket(mdl_request))
+    if (!is_transactional && clone_ticket(mdl_request))
     {
       /* Clone failed. */
       mdl_request->ticket= NULL;
@@ -1196,19 +1308,18 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request)
   if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
     return TRUE;
 
-  /* The below call also implicitly locks MDL_lock::m_mutex. */
+  /* The below call implicitly locks MDL_lock::m_rwlock on success. */
   if (!(lock= mdl_locks.find_or_insert(key)))
   {
     MDL_ticket::destroy(ticket);
     return TRUE;
   }
 
-  if (lock->can_grant_lock(this, mdl_request->type, FALSE))
+  if (lock->can_grant_lock(mdl_request->type, this))
   {
-    lock->granted.push_front(ticket);
-    pthread_mutex_unlock(&lock->m_mutex);
+    lock->m_granted.add_ticket(ticket);
+    rw_unlock(&lock->m_rwlock);
 
-    ticket->m_state= MDL_ACQUIRED;
     ticket->m_lock= lock;
 
     m_tickets.push_front(ticket);
@@ -1219,7 +1330,7 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request)
   {
     /* We can't get here if we allocated a new lock. */
     DBUG_ASSERT(! lock->is_empty());
-    pthread_mutex_unlock(&lock->m_mutex);
+    rw_unlock(&lock->m_rwlock);
     MDL_ticket::destroy(ticket);
   }
 
@@ -1228,42 +1339,10 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request)
 
 
 /**
-  Try to acquire one shared lock.
-
-  Unlike exclusive locks, shared locks are acquired one by
-  one. This is interface is chosen to simplify introduction of
-  the new locking API to the system. MDL_context::try_acquire_shared_lock()
-  is currently used from open_table(), and there we have only one
-  table to work with.
-
-  In future we may consider allocating multiple shared locks at once.
-
-  @param mdl_request [in/out] Lock request object for lock to be acquired
-
-  @retval  FALSE   Success. The lock may have not been acquired.
-                   Check the ticket, if it's NULL, a conflicting lock
-                   exists and another attempt should be made after releasing
-                   all current locks and waiting for conflicting lock go
-                   away (using MDL_context::wait_for_locks()).
-  @retval  TRUE    Out of resources, an error has been reported.
-*/
-
-bool
-MDL_context::try_acquire_shared_lock(MDL_request *mdl_request)
-{
-  DBUG_ASSERT(mdl_request->is_shared());
-  DBUG_ASSERT(mdl_request->type != MDL_SHARED_UPGRADABLE ||
-              is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
-
-  return try_acquire_lock_impl(mdl_request);
-}
-
-
-/**
-  Create a copy of a granted ticket. 
+  Create a copy of a granted ticket.
   This is used to make sure that HANDLER ticket
   is never shared with a ticket that belongs to
-  a transaction, so that when we HANDLER CLOSE, 
+  a transaction, so that when we HANDLER CLOSE,
   we don't release a transactional ticket, and
   vice versa -- when we COMMIT, we don't mistakenly
   release a ticket for an open HANDLER.
@@ -1278,25 +1357,30 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
   MDL_ticket *ticket;
 
   safe_mutex_assert_not_owner(&LOCK_open);
-  /* Only used for HANDLER. */
-  DBUG_ASSERT(mdl_request->ticket && mdl_request->ticket->is_shared());
-
+  /*
+    By submitting mdl_request->type to MDL_ticket::create()
+    we effectively downgrade the cloned lock to the level of
+    the request.
+  */
   if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
     return TRUE;
 
-  ticket->m_state= MDL_ACQUIRED;
+  /* clone() is not supposed to be used to get a stronger lock. */
+  DBUG_ASSERT(ticket->m_type <= mdl_request->ticket->m_type);
+
   ticket->m_lock= mdl_request->ticket->m_lock;
   mdl_request->ticket= ticket;
 
-  pthread_mutex_lock(&ticket->m_lock->m_mutex);
-  ticket->m_lock->granted.push_front(ticket);
-  pthread_mutex_unlock(&ticket->m_lock->m_mutex);
+  rw_wrlock(&ticket->m_lock->m_rwlock);
+  ticket->m_lock->m_granted.add_ticket(ticket);
+  rw_unlock(&ticket->m_lock->m_rwlock);
 
   m_tickets.push_front(ticket);
 
   return FALSE;
 }
 
+
 /**
   Notify a thread holding a shared metadata lock which
   conflicts with a pending exclusive lock.
@@ -1307,21 +1391,20 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
 
 void notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket)
 {
-  if (conflicting_ticket->is_shared())
+  /* Only try to abort locks on which we back off. */
+  if (conflicting_ticket->get_type() < MDL_SHARED_NO_WRITE)
   {
-    THD *conflicting_thd= conflicting_ticket->get_ctx()->get_thd();
+    MDL_context *conflicting_ctx= conflicting_ticket->get_ctx();
+    THD *conflicting_thd= conflicting_ctx->get_thd();
     DBUG_ASSERT(thd != conflicting_thd); /* Self-deadlock */
 
     /*
-      If the thread that holds the conflicting lock is waiting in MDL
-      subsystem it has to be woken up by calling MDL_context::awake().
-    */
-    conflicting_ticket->get_ctx()->awake();
-    /*
-      If it is waiting on table-level lock or some other non-MDL resource
-      we delegate its waking up to code outside of MDL.
+      If thread which holds conflicting lock is waiting on table-level
+      lock or some other non-MDL resource we might need to wake it up
+      by calling code outside of MDL.
     */
-    mysql_notify_thread_having_shared_lock(thd, conflicting_thd);
+    mysql_notify_thread_having_shared_lock(thd, conflicting_thd,
+                               conflicting_ctx->get_needs_thr_lock_abort());
   }
 }
 
@@ -1331,29 +1414,26 @@ void notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket)
 
   @param mdl_request  Request for the lock to be acqured.
 
-  @note Should not be used outside of MDL subsystem. Instead one should
-        call acquire_exclusive_lock() or acquire_exclusive_locks() methods
-        which ensure that conditions for deadlock-free lock acquisition are
-        fulfilled.
+  @note Should not be used outside of MDL subsystem. Instead one
+        should call acquire_lock() or acquire_locks()
+        methods which ensure that conditions for deadlock-free
+        lock acquisition are fulfilled.
 
   @retval FALSE  Success
   @retval TRUE   Failure
 */
 
-bool MDL_context::acquire_exclusive_lock_impl(MDL_request *mdl_request)
+bool MDL_context::acquire_lock_impl(MDL_request *mdl_request)
 {
   MDL_lock *lock;
-  const char *old_msg;
   MDL_ticket *ticket;
   bool not_used;
   st_my_thread_var *mysys_var= my_thread_var;
   MDL_key *key= &mdl_request->key;
 
-  DBUG_ASSERT(mdl_request->type == MDL_EXCLUSIVE &&
-              mdl_request->ticket == NULL);
-
   safe_mutex_assert_not_owner(&LOCK_open);
 
+  DBUG_ASSERT(mdl_request->ticket == NULL);
   /* Don't take chances in production. */
   mdl_request->ticket= NULL;
 
@@ -1363,129 +1443,67 @@ bool MDL_context::acquire_exclusive_lock_impl(MDL_request *mdl_request)
   */
   if ((ticket= find_ticket(mdl_request, &not_used)))
   {
-    DBUG_ASSERT(ticket->m_state == MDL_ACQUIRED);
-    DBUG_ASSERT(ticket->m_type == MDL_EXCLUSIVE);
+    DBUG_ASSERT(ticket->m_lock);
     mdl_request->ticket= ticket;
     return FALSE;
   }
 
-  DBUG_ASSERT(is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
+  DBUG_ASSERT(mdl_request->type < MDL_SHARED_NO_WRITE ||
+              is_lock_owner(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE));
 
   /* Early allocation: ticket will be needed in any case. */
   if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
     return TRUE;
 
-  /* The below call also implicitly locks MDL_lock::m_mutex. */
+  /* The below call implicitly locks MDL_lock::m_rwlock on success. */
   if (!(lock= mdl_locks.find_or_insert(key)))
   {
     MDL_ticket::destroy(ticket);
     return TRUE;
   }
 
-  lock->waiting_exclusive.push_front(ticket);
+  ticket->m_lock= lock;
 
-  old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_ctx_wakeup_cond,
-                          &lock->m_mutex);
+  lock->m_waiting.add_ticket(ticket);
 
-  while (!lock->can_grant_lock(this, mdl_request->type, FALSE))
+  while (!lock->can_grant_lock(mdl_request->type, this))
   {
-    if (m_lt_or_ha_sentinel)
-    {
-      /*
-        We're about to start waiting. Don't do it if we have
-        HANDLER locks (we can't have any other locks here).
-        Waiting with locks may lead to a deadlock.
+    wait_reset();
 
-        We have to do MDL_EXIT_COND here and then re-acquire the
-        lock as there is a chance that we will destroy MDL_lock
-        object and won't be able to call MDL_EXIT_COND after it.
-      */
-      MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
-
-      pthread_mutex_lock(&lock->m_mutex);
-      /* Get rid of pending ticket. */
-      lock->waiting_exclusive.remove(ticket);
-      if (lock->is_empty())
-        mdl_locks.remove(lock);
-      else
-      {
-        /*
-          There can be some contexts waiting to acquire shared
-          lock which now might be able to do it. Wake them up!
-        */
-        lock->wake_up_waiters();
-        pthread_mutex_unlock(&lock->m_mutex);
-      }
-      MDL_ticket::destroy(ticket);
-      my_error(ER_LOCK_DEADLOCK, MYF(0));
-      return TRUE;
-    }
+    if (ticket->is_upgradable_or_exclusive())
+      lock->notify_shared_locks(this);
 
-    MDL_ticket *conflicting_ticket;
-    MDL_lock::Ticket_iterator it(lock->granted);
+    rw_unlock(&lock->m_rwlock);
 
-    while ((conflicting_ticket= it++))
-      notify_shared_lock(m_thd, conflicting_ticket);
+    set_deadlock_weight(mdl_request->get_deadlock_weight());
+    will_wait_for(ticket);
 
     /* There is a shared or exclusive lock on the object. */
-    DEBUG_SYNC(m_thd, "mdl_acquire_exclusive_locks_wait");
+    DEBUG_SYNC(m_thd, "mdl_acquire_lock_wait");
 
-    /*
-      Another thread might have obtained a shared MDL lock on some table
-      but has not yet opened it and/or tried to obtain data lock on it.
-      Also invocation of acquire_exclusive_lock() method and consequently
-      first call to notify_shared_lock() might have happened right after
-      thread holding shared metadata lock in wait_for_locks() method
-      checked that there are no pending conflicting locks but before
-      it has started waiting.
-      In both these cases we need to sleep until these threads will start
-      waiting and try to abort them once again.
-
-      QQ: What is the optimal value for this sleep?
-    */
-    struct timespec abstime;
-    set_timespec(abstime, 1);
-    pthread_cond_timedwait(&m_ctx_wakeup_cond, &lock->m_mutex, &abstime);
+    bool is_deadlock= (find_deadlock() || timed_wait(1) == VICTIM_WAKE_UP);
 
-    if (mysys_var->abort)
-    {
-      /*
-        We have to do MDL_EXIT_COND here and then re-acquire the lock
-        as there is a chance that we will destroy MDL_lock object and
-        won't be able to call MDL_EXIT_COND after it.
-      */
-      MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+    stop_waiting();
 
-      pthread_mutex_lock(&lock->m_mutex);
-      /* Get rid of pending ticket. */
-      lock->waiting_exclusive.remove(ticket);
-      if (lock->is_empty())
-        mdl_locks.remove(lock);
-      else
-      {
-        /*
-          There can be some contexts waiting to acquire shared
-          lock which now might be able to do it. Wake them up!
-        */
-        lock->wake_up_waiters();
-        pthread_mutex_unlock(&lock->m_mutex);
-      }
+    if (is_deadlock || mysys_var->abort)
+    {
+      lock->remove_ticket(&MDL_lock::m_waiting, ticket);
       MDL_ticket::destroy(ticket);
+      if (is_deadlock)
+        my_error(ER_LOCK_DEADLOCK, MYF(0));
       return TRUE;
     }
+    rw_wrlock(&lock->m_rwlock);
   }
 
-  lock->waiting_exclusive.remove(ticket);
-  lock->granted.push_front(ticket);
+  lock->m_waiting.remove_ticket(ticket);
+  lock->m_granted.add_ticket(ticket);
 
-  if (lock->cached_object)
+  if (ticket->get_type() == MDL_EXCLUSIVE && lock->cached_object)
     (*lock->cached_object_release_hook)(lock->cached_object);
   lock->cached_object= NULL;
 
-  MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
-
-  ticket->m_state= MDL_ACQUIRED;
-  ticket->m_lock= lock;
+  rw_unlock(&lock->m_rwlock);
 
   m_tickets.push_front(ticket);
 
@@ -1495,28 +1513,6 @@ bool MDL_context::acquire_exclusive_lock_impl(MDL_request *mdl_request)
 }
 
 
-/**
-  Acquire an exclusive lock.
-
-  @param mdl_request  Request for the lock to be acqured.
-
-  @note Assumes that one already owns global intention exclusive lock.
-
-  @retval FALSE  Success
-  @retval TRUE   Failure
-*/
-
-bool MDL_context::acquire_exclusive_lock(MDL_request *mdl_request)
-{
-  /* Exclusive locks must always be acquired first, all at once. */
-  DBUG_ASSERT(! m_tickets.is_empty() &&
-              m_tickets.front()->m_lock->key.mdl_namespace() == MDL_key::GLOBAL &&
-              ++Ticket_list::Iterator(m_tickets) == m_lt_or_ha_sentinel);
-
-  return acquire_exclusive_lock_impl(mdl_request);
-}
-
-
 extern "C" int mdl_request_ptr_cmp(const void* ptr1, const void* ptr2)
 {
   MDL_request *req1= *(MDL_request**)ptr1;
@@ -1543,37 +1539,37 @@ extern "C" int mdl_request_ptr_cmp(const void* ptr1, const void* ptr2)
   @retval TRUE   Failure
 */
 
-bool MDL_context::acquire_exclusive_locks(MDL_request_list *mdl_requests)
+bool MDL_context::acquire_locks(MDL_request_list *mdl_requests)
 {
   MDL_request_list::Iterator it(*mdl_requests);
-  MDL_request **sort_buf;
-  uint i;
+  MDL_request **sort_buf, **p_req;
+  uint req_count= mdl_requests->elements();
+
+  if (req_count == 0)
+    return FALSE;
 
   /*
-    Exclusive locks must always be acquired first, all at once.
+    To reduce deadlocks, the server acquires all exclusive
+    locks at once. For shared locks, try_acquire_lock() is
+    used instead.
   */
-  DBUG_ASSERT(! m_tickets.is_empty() &&
-              m_tickets.front()->m_lock->key.mdl_namespace() == MDL_key::GLOBAL &&
-              ++Ticket_list::Iterator(m_tickets) == m_lt_or_ha_sentinel);
-
-  if (mdl_requests->is_empty())
-    return FALSE;
+  DBUG_ASSERT(m_tickets.is_empty() || m_tickets.front() == m_trans_sentinel);
 
   /* Sort requests according to MDL_key. */
-  if (! (sort_buf= (MDL_request **)my_malloc(mdl_requests->elements() *
-                                             sizeof(MDL_request *),
+  if (! (sort_buf= (MDL_request **)my_malloc(req_count *
+                                             sizeof(MDL_request*),
                                              MYF(MY_WME))))
     return TRUE;
 
-  for (i= 0; i < mdl_requests->elements(); i++)
-    sort_buf[i]= it++;
+  for (p_req= sort_buf; p_req < sort_buf + req_count; p_req++)
+    *p_req= it++;
 
-  my_qsort(sort_buf, mdl_requests->elements(), sizeof(MDL_request*),
+  my_qsort(sort_buf, req_count, sizeof(MDL_request*),
            mdl_request_ptr_cmp);
 
-  for (i= 0; i < mdl_requests->elements(); i++)
+  for (p_req= sort_buf; p_req < sort_buf + req_count; p_req++)
   {
-    if (acquire_exclusive_lock_impl(sort_buf[i]))
+    if (acquire_lock_impl(*p_req))
       goto err;
   }
   my_free(sort_buf, MYF(0));
@@ -1581,11 +1577,12 @@ bool MDL_context::acquire_exclusive_locks(MDL_request_list *mdl_requests)
 
 err:
   /* Release locks we have managed to acquire so far. */
-  for (i= 0; i < mdl_requests->elements() && sort_buf[i]->ticket; i++)
+  for (req_count= p_req - sort_buf, p_req= sort_buf;
+       p_req < sort_buf + req_count; p_req++)
   {
-    release_lock(sort_buf[i]->ticket);
+    release_lock((*p_req)->ticket);
     /* Reset lock request back to its initial state. */
-    sort_buf[i]->ticket= NULL;
+    (*p_req)->ticket= NULL;
   }
   my_free(sort_buf, MYF(0));
   return TRUE;
@@ -1612,251 +1609,198 @@ err:
 */
 
 bool
-MDL_ticket::upgrade_shared_lock_to_exclusive()
+MDL_context::upgrade_shared_lock_to_exclusive(MDL_ticket *mdl_ticket)
 {
-  const char *old_msg;
-  st_my_thread_var *mysys_var= my_thread_var;
-  THD *thd= m_ctx->get_thd();
-  MDL_ticket *pending_ticket;
+  MDL_request mdl_xlock_request;
+  MDL_ticket *mdl_svp= mdl_savepoint();
+  bool is_new_ticket;
 
   DBUG_ENTER("MDL_ticket::upgrade_shared_lock_to_exclusive");
-  DEBUG_SYNC(thd, "mdl_upgrade_shared_lock_to_exclusive");
+  DEBUG_SYNC(get_thd(), "mdl_upgrade_shared_lock_to_exclusive");
 
-  safe_mutex_assert_not_owner(&LOCK_open);
-
-  /* Allow this function to be called twice for the same lock request. */
-  if (m_type == MDL_EXCLUSIVE)
+  /*
+    Do nothing if already upgraded. Used when we FLUSH TABLE under
+    LOCK TABLES and a table is listed twice in LOCK TABLES list.
+  */
+  if (mdl_ticket->m_type == MDL_EXCLUSIVE)
     DBUG_RETURN(FALSE);
 
-  /* Only allow upgrades from MDL_SHARED_UPGRADABLE */
-  DBUG_ASSERT(m_type == MDL_SHARED_UPGRADABLE);
+  /* Only allow upgrades from MDL_SHARED_NO_WRITE/NO_READ_WRITE */
+  DBUG_ASSERT(mdl_ticket->m_type == MDL_SHARED_NO_WRITE ||
+              mdl_ticket->m_type == MDL_SHARED_NO_READ_WRITE);
 
-  /*
-    Since we should have already acquired an intention exclusive
-    global lock this call is only enforcing asserts.
-  */
-  DBUG_ASSERT(m_ctx->is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
+  mdl_xlock_request.init(&mdl_ticket->m_lock->key, MDL_EXCLUSIVE);
 
-  /*
-    Create an auxiliary ticket to represent a pending exclusive
-    lock and add it to the 'waiting' queue for the duration
-    of upgrade. During upgrade we abort waits of connections
-    that own conflicting locks. A pending request is used
-    to signal such connections that upon waking up they
-    must back off, rather than fall into sleep again.
-  */
-  if (! (pending_ticket= MDL_ticket::create(m_ctx, MDL_EXCLUSIVE)))
+  if (acquire_lock_impl(&mdl_xlock_request))
     DBUG_RETURN(TRUE);
 
-  pthread_mutex_lock(&m_lock->m_mutex);
+  is_new_ticket= ! has_lock(mdl_svp, mdl_xlock_request.ticket);
 
-  m_lock->waiting_exclusive.push_front(pending_ticket);
+  /* Merge the acquired and the original lock. @todo: move to a method. */
+  rw_wrlock(&mdl_ticket->m_lock->m_rwlock);
+  if (is_new_ticket)
+    mdl_ticket->m_lock->m_granted.remove_ticket(mdl_xlock_request.ticket);
+  /*
+    Set the new type of lock in the ticket. To update state of
+    MDL_lock object correctly we need to temporarily exclude
+    ticket from the granted queue and then include it back.
+  */
+  mdl_ticket->m_lock->m_granted.remove_ticket(mdl_ticket);
+  mdl_ticket->m_type= MDL_EXCLUSIVE;
+  mdl_ticket->m_lock->m_granted.add_ticket(mdl_ticket);
 
-  old_msg= MDL_ENTER_COND(thd, mysys_var, &m_ctx->m_ctx_wakeup_cond,
-                          &m_lock->m_mutex);
+  rw_unlock(&mdl_ticket->m_lock->m_rwlock);
 
-  while (1)
+  if (is_new_ticket)
   {
-    if (m_lock->can_grant_lock(m_ctx, MDL_EXCLUSIVE, TRUE))
-      break;
+    m_tickets.remove(mdl_xlock_request.ticket);
+    MDL_ticket::destroy(mdl_xlock_request.ticket);
+  }
 
-    MDL_ticket *conflicting_ticket;
-    MDL_lock::Ticket_iterator it(m_lock->granted);
+  DBUG_RETURN(FALSE);
+}
 
-    /*
-      If m_ctx->lt_or_ha_sentinel(), and this sentinel is for HANDLER,
-      we can deadlock. However, HANDLER is not allowed under
-      LOCK TABLES, and apart from LOCK TABLES there are only
-      two cases of lock upgrade: ALTER TABLE and CREATE/DROP
-      TRIGGER (*). This leaves us with the following scenario
-      for deadlock:
-
-      connection 1                          connection 2
-      handler t1 open;                      handler t2 open;
-      alter table t2 ...                    alter table t1 ...
-
-      This scenario is quite remote, since ALTER
-      (and CREATE/DROP TRIGGER) performs mysql_ha_flush() in
-      the beginning, and thus closes open HANDLERS against which
-      there is a pending lock upgrade. Still, two ALTER statements
-      can interleave and not notice each other's pending lock
-      (e.g. if both upgrade their locks at the same time).
-      This, however, is quite unlikely, so we do nothing to
-      address it.
-
-      (*) There is no requirement to upgrade lock in
-      CREATE/DROP TRIGGER, it's used there just for convenience.
-
-      A temporary work-around to avoid deadlocks/livelocks in
-      a situation when in one connection ALTER TABLE tries to
-      upgrade its metadata lock and in another connection
-      the active transaction already got this lock in some
-      of its earlier statements.
-      In such case this transaction always succeeds with getting
-      a metadata lock on the table -- it already has one.
-      But later on it may block on the table level lock, since ALTER
-      got TL_WRITE_ALLOW_READ, and subsequently get aborted
-      by notify_shared_lock().
-      An abort will lead to a back off, and a second attempt to
-      get an MDL lock (successful), and a table lock (-> livelock).
-
-      The call below breaks this loop by forcing transactions to call
-      tdc_wait_for_old_versions() (even if the transaction doesn't need
-      any new metadata locks), which in turn will check if someone
-      is waiting on the owned MDL lock, and produce ER_LOCK_DEADLOCK.
-
-      TODO: Long-term such deadlocks/livelock will be resolved within
-            MDL subsystem and thus this call will become unnecessary.
-    */
-    mysql_abort_transactions_with_shared_lock(&m_lock->key);
 
-    while ((conflicting_ticket= it++))
-    {
-      if (conflicting_ticket->m_ctx != m_ctx)
-        notify_shared_lock(thd, conflicting_ticket);
-    }
+bool MDL_lock::find_deadlock(MDL_ticket *waiting_ticket,
+                             Deadlock_detection_context *deadlock_ctx)
+{
+  MDL_ticket *ticket;
+  bool result= FALSE;
 
-    /* There is a shared or exclusive lock on the object. */
-    DEBUG_SYNC(thd, "mdl_upgrade_shared_lock_to_exclusive_wait");
+  rw_rdlock(&m_rwlock);
 
-    /*
-      Another thread might have obtained a shared MDL lock on some table
-      but has not yet opened it and/or tried to obtain data lock on it.
-      Also invocation of acquire_exclusive_lock() method and consequently
-      first call to notify_shared_lock() might have happened right after
-      thread holding shared metadata lock in wait_for_locks() method
-      checked that there are no pending conflicting locks but before
-      it has started waiting.
-      In both these cases we need to sleep until these threads will start
-      waiting and try to abort them once again.
-    */
-    struct timespec abstime;
-    set_timespec(abstime, 1);
-    pthread_cond_timedwait(&m_ctx->m_ctx_wakeup_cond, &m_lock->m_mutex,
-                           &abstime);
+  Ticket_iterator granted_it(m_granted);
+  Ticket_iterator waiting_it(m_waiting);
 
-    if (mysys_var->abort)
+  while ((ticket= granted_it++))
+  {
+    if (ticket->is_incompatible_when_granted(waiting_ticket->get_type()) &&
+        ticket->get_ctx() != waiting_ticket->get_ctx() &&
+        ticket->get_ctx() == deadlock_ctx->start)
     {
-      m_lock->waiting_exclusive.remove(pending_ticket);
-      /*
-        If there are no other pending requests for exclusive locks
-        we need to wake up threads waiting for a chance to acquire
-        shared lock.
-      */
-      m_lock->wake_up_waiters();
-      MDL_EXIT_COND(thd, mysys_var, &m_lock->m_mutex, old_msg);
-      MDL_ticket::destroy(pending_ticket);
-      DBUG_RETURN(TRUE);
+      result= TRUE;
+      goto end;
     }
   }
 
-  /* Set the new type of lock in the ticket. */
-  m_type= MDL_EXCLUSIVE;
-
-  /* Remove and destroy the auxiliary pending ticket. */
-  m_lock->waiting_exclusive.remove(pending_ticket);
-
-  if (m_lock->cached_object)
-    (*m_lock->cached_object_release_hook)(m_lock->cached_object);
-  m_lock->cached_object= 0;
+  while ((ticket= waiting_it++))
+  {
+    if (ticket->is_incompatible_when_waiting(waiting_ticket->get_type()) &&
+        ticket->get_ctx() != waiting_ticket->get_ctx() &&
+        ticket->get_ctx() == deadlock_ctx->start)
+    {
+      result= TRUE;
+      goto end;
+    }
+  }
 
-  MDL_EXIT_COND(thd, mysys_var, &m_lock->m_mutex, old_msg);
+  granted_it.rewind();
+  while ((ticket= granted_it++))
+  {
+    if (ticket->is_incompatible_when_granted(waiting_ticket->get_type()) &&
+        ticket->get_ctx() != waiting_ticket->get_ctx() &&
+        ticket->get_ctx()->find_deadlock(deadlock_ctx))
+    {
+      result= TRUE;
+      goto end;
+    }
+  }
 
-  MDL_ticket::destroy(pending_ticket);
+  waiting_it.rewind();
+  while ((ticket= waiting_it++))
+  {
+    if (ticket->is_incompatible_when_waiting(waiting_ticket->get_type()) &&
+        ticket->get_ctx() != waiting_ticket->get_ctx() &&
+        ticket->get_ctx()->find_deadlock(deadlock_ctx))
+    {
+      result= TRUE;
+      goto end;
+    }
+  }
 
-  DBUG_RETURN(FALSE);
+end:
+  rw_unlock(&m_rwlock);
+  return result;
 }
 
 
-/**
-  Try to acquire an exclusive lock on the object if there are
-  no conflicting locks.
-
-  Similar to the previous function, but returns
-  immediately without any side effect if encounters a lock
-  conflict. Otherwise takes the lock.
-
-  This function is used in CREATE TABLE ... LIKE to acquire a lock
-  on the table to be created. In this statement we don't want to
-  block and wait for the lock if the table already exists.
-
-  @param mdl_request [in] The lock request
-  @param conflict   [out] Indicates that conflicting lock exists
-
-  @retval TRUE  Failure: some error occurred (probably OOM).
-  @retval FALSE Success: the lock might have not been acquired,
-                         check request.ticket to find out.
-
-  FIXME: Compared to lock_table_name_if_not_cached()
-         it gives slightly more false negatives.
-*/
-
-bool
-MDL_context::try_acquire_exclusive_lock(MDL_request *mdl_request)
+bool MDL_context::find_deadlock(Deadlock_detection_context *deadlock_ctx)
 {
-  DBUG_ASSERT(mdl_request->type == MDL_EXCLUSIVE);
-  DBUG_ASSERT(is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
+  bool result= FALSE;
 
-  return try_acquire_lock_impl(mdl_request);
-}
+  rw_rdlock(&m_waiting_for_lock);
 
+  if (m_waiting_for)
+  {
+    /*
+      QQ: should we rather be checking for NO_WAKE_UP ?
 
-/**
-  Acquire the global shared metadata lock.
-
-  Holding this lock will block all requests for exclusive locks
-  and shared locks which can be potentially upgraded to exclusive.
-
-  @retval FALSE Success -- the lock was granted.
-  @retval TRUE  Failure -- our thread was killed.
-*/
-
-bool MDL_context::acquire_global_shared_lock()
-{
-  MDL_request mdl_request;
-
-  DBUG_ASSERT(! is_global_lock_owner(MDL_SHARED));
-
-  mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED);
+      We want to do check signal only when m_waiting_for is set
+      to avoid reading left-overs from previous kills.
+    */
+    if (peek_signal() != VICTIM_WAKE_UP)
+    {
 
-  if (acquire_lock_impl(&mdl_request))
-    return TRUE;
+      if (++deadlock_ctx->current_search_depth >
+          deadlock_ctx->MAX_SEARCH_DEPTH)
+        result= TRUE;
+      else
+        result= m_waiting_for->m_lock->find_deadlock(m_waiting_for,
+                                                     deadlock_ctx);
+      --deadlock_ctx->current_search_depth;
+    }
+  }
 
-  move_ticket_after_lt_or_ha_sentinel(mdl_request.ticket);
+  if (result)
+  {
+    if (! deadlock_ctx->victim)
+      deadlock_ctx->victim= this;
+    else if (deadlock_ctx->victim->m_deadlock_weight >= m_deadlock_weight)
+    {
+      rw_unlock(&deadlock_ctx->victim->m_waiting_for_lock);
+      deadlock_ctx->victim= this;
+    }
+    else
+      rw_unlock(&m_waiting_for_lock);
+  }
+  else
+    rw_unlock(&m_waiting_for_lock);
 
-  return FALSE;
+  return result;
 }
 
 
-/**
-  Implement a simple deadlock detection heuristic: check if there
-  are any pending exclusive locks which conflict with shared locks
-  held by this thread. In that case waiting can be circular,
-  i.e. lead to a deadlock.
-
-  @return TRUE   If there are any pending conflicting locks.
-          FALSE  Otherwise.
-*/
-
-bool MDL_context::can_wait_lead_to_deadlock() const
+bool MDL_context::find_deadlock()
 {
-  Ticket_iterator ticket_it(m_tickets);
-  MDL_ticket *ticket;
+  Deadlock_detection_context deadlock_ctx(this);
 
-  while ((ticket= ticket_it++))
+  while (1)
   {
-    /*
-      In MySQL we never call this method while holding exclusive or
-      upgradeable shared metadata locks.
-      Otherwise we would also have to check for the presence of pending
-      requests for conflicting types of global lock.
-      In addition MDL_ticket::has_pending_conflicting_lock()
-      won't work properly for exclusive type of lock.
-    */
-    DBUG_ASSERT(! ticket->is_upgradable_or_exclusive());
+    if (! find_deadlock(&deadlock_ctx))
+    {
+      /* No deadlocks are found! */
+      break;
+    }
 
-    if (ticket->has_pending_conflicting_lock())
+    if (deadlock_ctx.victim != this)
+    {
+      deadlock_ctx.victim->awake(VICTIM_WAKE_UP);
+      rw_unlock(&deadlock_ctx.victim->m_waiting_for_lock);
+      /*
+        After adding new arc to waiting graph we found that it participates
+        in some loop (i.e. there is a deadlock). We decided to destroy this
+        loop by removing some arc other than newly added. Since this doesn't
+        guarantee that all loops created by addition of this arc are
+        destroyed we have to repeat search.
+      */
+      continue;
+    }
+    else
+    {
+      DBUG_ASSERT(&deadlock_ctx.victim->m_waiting_for_lock == &m_waiting_for_lock);
+      rw_unlock(&deadlock_ctx.victim->m_waiting_for_lock);
       return TRUE;
+    }
   }
   return FALSE;
 }
@@ -1876,16 +1820,15 @@ bool MDL_context::can_wait_lead_to_deadlock() const
 */
 
 bool
-MDL_context::wait_for_locks(MDL_request_list *mdl_requests)
+MDL_context::wait_for_lock(MDL_request *mdl_request)
 {
   MDL_lock *lock;
-  MDL_request *mdl_request;
-  MDL_request_list::Iterator it(*mdl_requests);
-  const char *old_msg;
   st_my_thread_var *mysys_var= my_thread_var;
 
   safe_mutex_assert_not_owner(&LOCK_open);
 
+  DBUG_ASSERT(mdl_request->ticket == NULL);
+
   while (!mysys_var->abort)
   {
     /*
@@ -1900,87 +1843,45 @@ MDL_context::wait_for_locks(MDL_request_list *mdl_requests)
     */
     mysql_ha_flush(m_thd);
 
-    /*
-      In cases when we wait while still holding some metadata
-      locks deadlocks are possible.
-      To avoid them we use the following simple empiric - don't
-      wait for new lock request to be satisfied if for one of the
-      locks which are already held by this connection there is
-      a conflicting request (i.e. this connection should not wait
-      if someone waits for it).
-      This empiric should work well (e.g. give low number of false
-      negatives) in situations when conflicts are rare (in our
-      case this is true since DDL statements should be rare).
-    */
-    if (can_wait_lead_to_deadlock())
+    MDL_key *key= &mdl_request->key;
+
+    /* The below call implicitly locks MDL_lock::m_rwlock on success. */
+    if (! (lock= mdl_locks.find(key)))
+      return FALSE;
+
+    if (lock->can_grant_lock(mdl_request->type, this))
     {
-      my_error(ER_LOCK_DEADLOCK, MYF(0));
-      return TRUE;
+      rw_unlock(&lock->m_rwlock);
+      return FALSE;
     }
 
-    it.rewind();
-    while ((mdl_request= it++))
+    MDL_ticket *pending_ticket;
+    if (! (pending_ticket= MDL_ticket::create(this, mdl_request->type)))
     {
-      MDL_key *key= &mdl_request->key;
-      DBUG_ASSERT(mdl_request->ticket == NULL);
-
-      /*
-        To avoid starvation we don't wait if we have a conflict against
-        request for MDL_EXCLUSIVE lock.
-      */
-      if (mdl_request->is_shared() ||
-          mdl_request->type == MDL_INTENTION_EXCLUSIVE)
-      {
-        /* The below call also implicitly locks MDL_lock::m_mutex. */
-        if (! (lock= mdl_locks.find(key)))
-          continue;
-
-        if (lock->can_grant_lock(this, mdl_request->type, FALSE))
-        {
-          pthread_mutex_unlock(&lock->m_mutex);
-          continue;
-        }
-
-        MDL_ticket *pending_ticket;
-        if (! (pending_ticket= MDL_ticket::create(this, mdl_request->type)))
-        {
-          pthread_mutex_unlock(&lock->m_mutex);
-          return TRUE;
-        }
-        if (mdl_request->is_shared())
-          lock->waiting_shared.push_front(pending_ticket);
-        else
-          lock->waiting_exclusive.push_front(pending_ticket);
-
-        old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_ctx_wakeup_cond,
-                                &lock->m_mutex);
-
-        pthread_cond_wait(&m_ctx_wakeup_cond, &lock->m_mutex);
-
-        /*
-          We have to do MDL_EXIT_COND here and then re-acquire the lock
-          as there is a chance that we will destroy MDL_lock object and
-          won't be able to call MDL_EXIT_COND after it.
-        */
-        MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
-
-        pthread_mutex_lock(&lock->m_mutex);
-        if (mdl_request->is_shared())
-          lock->waiting_shared.remove(pending_ticket);
-        else
-          lock->waiting_exclusive.remove(pending_ticket);
-        if (lock->is_empty())
-          mdl_locks.remove(lock);
-        else
-          pthread_mutex_unlock(&lock->m_mutex);
-        MDL_ticket::destroy(pending_ticket);
-        break;
-      }
+      rw_unlock(&lock->m_rwlock);
+      return TRUE;
     }
-    if (!mdl_request)
+
+    pending_ticket->m_lock= lock;
+
+    lock->m_waiting.add_ticket(pending_ticket);
+
+    wait_reset();
+    rw_unlock(&lock->m_rwlock);
+
+    set_deadlock_weight(MDL_DEADLOCK_WEIGHT_DML);
+    will_wait_for(pending_ticket);
+
+    bool is_deadlock= (find_deadlock() || wait() == VICTIM_WAKE_UP);
+
+    stop_waiting();
+
+    lock->remove_ticket(&MDL_lock::m_waiting, pending_ticket);
+    MDL_ticket::destroy(pending_ticket);
+    if (is_deadlock)
     {
-      /* There are no conflicts for any locks! */
-      break;
+      my_error(ER_LOCK_DEADLOCK, MYF(0));
+      return TRUE;
     }
   }
   return mysys_var->abort;
@@ -2000,23 +1901,13 @@ void MDL_context::release_lock(MDL_ticket *ticket)
   DBUG_PRINT("enter", ("db=%s name=%s", lock->key.db_name(),
                                         lock->key.name()));
 
-  DBUG_ASSERT(this == ticket->m_ctx);
+  DBUG_ASSERT(this == ticket->get_ctx());
   safe_mutex_assert_not_owner(&LOCK_open);
 
-  if (ticket == m_lt_or_ha_sentinel)
-    m_lt_or_ha_sentinel= ++Ticket_list::Iterator(m_tickets, ticket);
-
-  pthread_mutex_lock(&lock->m_mutex);
+  if (ticket == m_trans_sentinel)
+    m_trans_sentinel= ++Ticket_list::Iterator(m_tickets, ticket);
 
-  lock->granted.remove(ticket);
-
-  if (lock->is_empty())
-    mdl_locks.remove(lock);
-  else
-  {
-    lock->wake_up_waiters();
-    pthread_mutex_unlock(&lock->m_mutex);
-  }
+  lock->remove_ticket(&MDL_lock::m_granted, ticket);
 
   m_tickets.remove(ticket);
   MDL_ticket::destroy(ticket);
@@ -2086,7 +1977,7 @@ void MDL_context::release_all_locks_for_name(MDL_ticket *name)
 
   while ((ticket= it_ticket++))
   {
-    DBUG_ASSERT(ticket->m_state == MDL_ACQUIRED);
+    DBUG_ASSERT(ticket->m_lock);
     /*
       We rarely have more than one ticket in this loop,
       let's not bother saving on pthread_cond_broadcast().
@@ -2099,89 +1990,43 @@ void MDL_context::release_all_locks_for_name(MDL_ticket *name)
 
 /**
   Downgrade an exclusive lock to shared metadata lock.
-*/
-
-void MDL_ticket::downgrade_exclusive_lock()
-{
-  safe_mutex_assert_not_owner(&LOCK_open);
-
-  if (is_shared())
-    return;
-
-  pthread_mutex_lock(&m_lock->m_mutex);
-  m_type= MDL_SHARED_UPGRADABLE;
-
-  if (! m_lock->waiting_shared.is_empty())
-  {
-    MDL_lock::Ticket_iterator it(m_lock->waiting_shared);
-    MDL_ticket *ticket;
-    while ((ticket= it++))
-      ticket->get_ctx()->awake();
-  }
-
-  pthread_mutex_unlock(&m_lock->m_mutex);
-}
 
-
-/**
-  Release the global shared metadata lock.
+  @param type  Type of lock to which exclusive lock should be downgraded.
 */
 
-void MDL_context::release_global_shared_lock()
+void MDL_ticket::downgrade_exclusive_lock(enum_mdl_type type)
 {
-  MDL_request mdl_request;
-  MDL_ticket *ticket;
-  bool not_used;
-
-  mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED);
-
   safe_mutex_assert_not_owner(&LOCK_open);
 
   /*
-    TODO/QQ/FIXME: In theory we always should be able to find
-                   ticket here. But in practice this is not
-                   always TRUE.
+    Do nothing if already downgraded. Used when we FLUSH TABLE under
+    LOCK TABLES and a table is listed twice in LOCK TABLES list.
   */
+  if (m_type != MDL_EXCLUSIVE)
+    return;
 
-  if ((ticket= find_ticket(&mdl_request, &not_used)))
-    release_lock(ticket);
-}
-
-
-/**
-  Auxiliary function which allows to check if we have exclusive lock
-  on the object.
-
-  @param mdl_namespace Id of object namespace
-  @param db            Name of the database
-  @param name          Name of the object
-
-  @return TRUE if current context contains exclusive lock for the object,
-          FALSE otherwise.
-*/
-
-bool
-MDL_context::is_exclusive_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
-                                     const char *db, const char *name)
-{
-  MDL_request mdl_request;
-  bool is_lt_or_ha_unused;
-  mdl_request.init(mdl_namespace, db, name, MDL_EXCLUSIVE);
-  MDL_ticket *ticket= find_ticket(&mdl_request, &is_lt_or_ha_unused);
-
-  DBUG_ASSERT(ticket == NULL || ticket->m_state == MDL_ACQUIRED);
-
-  return ticket;
+  rw_wrlock(&m_lock->m_rwlock);
+  /*
+    To update state of MDL_lock object correctly we need to temporarily
+    exclude ticket from the granted queue and then include it back.
+  */
+  m_lock->m_granted.remove_ticket(this);
+  m_type= type;
+  m_lock->m_granted.add_ticket(this);
+  m_lock->wake_up_waiters();
+  rw_unlock(&m_lock->m_rwlock);
 }
 
 
 /**
   Auxiliary function which allows to check if we have some kind of lock on
-  a object.
+  a object. Returns TRUE if we have a lock of a given or stronger type.
 
   @param mdl_namespace Id of object namespace
   @param db            Name of the database
   @param name          Name of the object
+  @param mdl_type      Lock type. Pass in the weakest type to find
+                       out if there is at least some lock.
 
   @return TRUE if current context contains satisfied lock for the object,
           FALSE otherwise.
@@ -2189,25 +2034,22 @@ MDL_context::is_exclusive_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
 
 bool
 MDL_context::is_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
-                           const char *db, const char *name)
+                           const char *db, const char *name,
+                           enum_mdl_type mdl_type)
 {
-  MDL_key key(mdl_namespace, db, name);
-  MDL_ticket *ticket;
-  MDL_context::Ticket_iterator it(m_tickets);
+  MDL_request mdl_request;
+  bool is_transactional_unused;
+  mdl_request.init(mdl_namespace, db, name, mdl_type);
+  MDL_ticket *ticket= find_ticket(&mdl_request, &is_transactional_unused);
 
-  while ((ticket= it++))
-  {
-    if (ticket->m_lock->key.is_equal(&key))
-      break;
-  }
+  DBUG_ASSERT(ticket == NULL || ticket->m_lock);
 
   return ticket;
 }
 
 
 /**
-  Check if we have any pending exclusive locks which conflict with
-  existing shared lock.
+  Check if we have any pending locks which conflict with existing shared lock.
 
   @pre The ticket must match an acquired lock.
 
@@ -2216,10 +2058,7 @@ MDL_context::is_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
 
 bool MDL_ticket::has_pending_conflicting_lock() const
 {
-  safe_mutex_assert_not_owner(&LOCK_open);
-  DBUG_ASSERT(is_shared());
-
-  return m_lock->has_pending_exclusive_lock();
+  return m_lock->has_pending_conflicting_lock(m_type);
 }
 
 
@@ -2304,7 +2143,7 @@ void MDL_context::rollback_to_savepoint(MDL_ticket *mdl_savepoint)
 
   /* If savepoint is NULL, it is from the start of the transaction. */
   release_locks_stored_before(mdl_savepoint ?
-                              mdl_savepoint : m_lt_or_ha_sentinel);
+                              mdl_savepoint : m_trans_sentinel);
 
   DBUG_VOID_RETURN;
 }
@@ -2322,7 +2161,7 @@ void MDL_context::rollback_to_savepoint(MDL_ticket *mdl_savepoint)
 void MDL_context::release_transactional_locks()
 {
   DBUG_ENTER("MDL_context::release_transactional_locks");
-  release_locks_stored_before(m_lt_or_ha_sentinel);
+  release_locks_stored_before(m_trans_sentinel);
   DBUG_VOID_RETURN;
 }
 
@@ -2331,20 +2170,21 @@ void MDL_context::release_transactional_locks()
   Does this savepoint have this lock?
   
   @retval TRUE  The ticket is older than the savepoint and
-                is not LT or HA ticket. Thus it belongs to
-                the savepoint.
+                is not LT, HA or GLR ticket. Thus it belongs
+                to the savepoint.
   @retval FALSE The ticket is newer than the savepoint
-                or is an LT or HA ticket.
+                or is an LT, HA or GLR ticket.
 */
 
 bool MDL_context::has_lock(MDL_ticket *mdl_savepoint,
                            MDL_ticket *mdl_ticket)
 {
   MDL_ticket *ticket;
+  /* Start from the beginning, most likely mdl_ticket's been just acquired. */
   MDL_context::Ticket_iterator it(m_tickets);
   bool found_savepoint= FALSE;
 
-  while ((ticket= it++) && ticket != m_lt_or_ha_sentinel)
+  while ((ticket= it++) && ticket != m_trans_sentinel)
   {
     /*
       First met the savepoint. The ticket must be
@@ -2359,28 +2199,28 @@ bool MDL_context::has_lock(MDL_ticket *mdl_savepoint,
     if (ticket == mdl_ticket)
       return found_savepoint;
   }
-  /* Reached m_lt_or_ha_sentinel. The ticket must be an LT or HA ticket. */
+  /* Reached m_trans_sentinel. The ticket must be LT, HA or GRL ticket. */
   return FALSE;
 }
 
 
 /**
   Rearrange the ticket to reside in the part of the list that's
-  beyond m_lt_or_ha_sentinel. This effectively changes the ticket
+  beyond m_trans_sentinel. This effectively changes the ticket
   life cycle, from automatic to manual: i.e. the ticket is no
   longer released by MDL_context::release_transactional_locks() or
   MDL_context::rollback_to_savepoint(), it must be released manually.
 */
 
-void MDL_context::move_ticket_after_lt_or_ha_sentinel(MDL_ticket *mdl_ticket)
+void MDL_context::move_ticket_after_trans_sentinel(MDL_ticket *mdl_ticket)
 {
   m_tickets.remove(mdl_ticket);
-  if (m_lt_or_ha_sentinel == NULL)
+  if (m_trans_sentinel == NULL)
   {
-    m_lt_or_ha_sentinel= mdl_ticket;
+    m_trans_sentinel= mdl_ticket;
     /* sic: linear from the number of transactional tickets acquired so-far! */
     m_tickets.push_back(mdl_ticket);
   }
   else
-    m_tickets.insert_after(m_lt_or_ha_sentinel, mdl_ticket);
+    m_tickets.insert_after(m_trans_sentinel, mdl_ticket);
 }
diff --git a/sql/mdl.h b/sql/mdl.h
index 3ae7cdc743d..848c446496f 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -26,28 +26,121 @@ class THD;
 class MDL_context;
 class MDL_lock;
 class MDL_ticket;
+class Deadlock_detection_context;
 
 /**
   Type of metadata lock request.
 
-  - High-priority shared locks differ from ordinary shared locks by
-    that they ignore pending requests for exclusive locks.
-  - Upgradable shared locks can be later upgraded to exclusive
-    (because of that their acquisition involves implicit
-     acquisition of global intention-exclusive lock).
-
   @sa Comments for MDL_object_lock::can_grant_lock() and
-  MDL_global_lock::can_grant_lock() for details.
+      MDL_global_lock::can_grant_lock() for details.
 */
 
-enum enum_mdl_type {MDL_SHARED=0, MDL_SHARED_HIGH_PRIO,
-                    MDL_SHARED_UPGRADABLE, MDL_INTENTION_EXCLUSIVE,
-                    MDL_EXCLUSIVE};
-
-
-/** States which a metadata lock ticket can have. */
+enum enum_mdl_type {
+  /*
+    An intention exclusive metadata lock. Used only for global locks.
+    Owner of this type of lock can acquire upgradable exclusive locks on
+    individual objects.
+    Compatible with other IX locks, but is incompatible with global S lock.
+  */
+  MDL_INTENTION_EXCLUSIVE= 0,
+  /*
+    A shared metadata lock.
+    To be used in cases when we are interested in object metadata only
+    and there is no intention to access object data (e.g. for stored
+    routines or during preparing prepared statements).
+    We also mis-use this type of lock for open HANDLERs, since lock
+    acquired by this statement has to be compatible with lock acquired
+    by LOCK TABLES ... WRITE statement, i.e. SNRW (We can't get by by
+    acquiring S lock at HANDLER ... OPEN time and upgrading it to SR
+    lock for HANDLER ... READ as it doesn't solve problem with need
+    to abort DML statements which wait on table level lock while having
+    open HANDLER in the same connection).
+    To avoid deadlock which may occur when SNRW lock is being upgraded to
+    X lock for table on which there is an active S lock which is owned by
+    thread which waits in its turn for table-level lock owned by thread
+    performing upgrade we have to use thr_abort_locks_for_thread()
+    facility in such situation.
+    This problem does not arise for locks on stored routines as we don't
+    use SNRW locks for them. It also does not arise when S locks are used
+    during PREPARE calls as table-level locks are not acquired in this
+    case.
+  */
+  MDL_SHARED,
+  /*
+    A high priority shared metadata lock.
+    Used for cases when there is no intention to access object data (i.e.
+    data in the table).
+    "High priority" means that, unlike other shared locks, it is granted
+    ignoring pending requests for exclusive locks. Intended for use in
+    cases when we only need to access metadata and not data, e.g. when
+    filling an INFORMATION_SCHEMA table.
+    Since SH lock is compatible with SNRW lock, the connection that
+    holds SH lock lock should not try to acquire any kind of table-level
+    or row-level lock, as this can lead to a deadlock. Moreover, after
+    acquiring SH lock, the connection should not wait for any other
+    resource, as it might cause starvation for X locks and a potential
+    deadlock during upgrade of SNW or SNRW to X lock (e.g. if the
+    upgrading connection holds the resource that is being waited for).
+  */
+  MDL_SHARED_HIGH_PRIO,
+  /*
+    A shared metadata lock for cases when there is an intention to read data
+    from table.
+    A connection holding this kind of lock can read table metadata and read
+    table data (after acquiring appropriate table and row-level locks).
+    This means that one can only acquire TL_READ, TL_READ_NO_INSERT, and
+    similar table-level locks on table if one holds SR MDL lock on it.
+    To be used for tables in SELECTs, subqueries, and LOCK TABLE ...  READ
+    statements.
+  */
+  MDL_SHARED_READ,
+  /*
+    A shared metadata lock for cases when there is an intention to modify
+    (and not just read) data in the table.
+    A connection holding SW lock can read table metadata and modify or read
+    table data (after acquiring appropriate table and row-level locks).
+    To be used for tables to be modified by INSERT, UPDATE, DELETE
+    statements, but not LOCK TABLE ... WRITE or DDL). Also taken by
+    SELECT ... FOR UPDATE.
+  */
+  MDL_SHARED_WRITE,
+  /*
+    An upgradable shared metadata lock which blocks all attempts to update
+    table data, allowing reads.
+    A connection holding this kind of lock can read table metadata and read
+    table data.
+    Can be upgraded to X metadata lock.
+    Note, that since this type of lock is not compatible with SNRW or SW
+    lock types, acquiring appropriate engine-level locks for reading
+    (TL_READ* for MyISAM, shared row locks in InnoDB) should be
+    contention-free.
+    To be used for the first phase of ALTER TABLE, when copying data between
+    tables, to allow concurrent SELECTs from the table, but not UPDATEs.
+  */
+  MDL_SHARED_NO_WRITE,
+  /*
+    An upgradable shared metadata lock which allows other connections
+    to access table metadata, but not data.
+    It blocks all attempts to read or update table data, while allowing
+    INFORMATION_SCHEMA and SHOW queries.
+    A connection holding this kind of lock can read table metadata modify and
+    read table data.
+    Can be upgraded to X metadata lock.
+    To be used for LOCK TABLES WRITE statement.
+    Not compatible with any other lock type except S and SH.
+  */
+  MDL_SHARED_NO_READ_WRITE,
+  /*
+    An exclusive metadata lock.
+    A connection holding this lock can modify both table's metadata and data.
+    No other type of metadata lock can be granted while this lock is held.
+    To be used for CREATE/DROP/RENAME TABLE statements and for execution of
+    certain phases of other DDL statements.
+  */
+  MDL_EXCLUSIVE,
+  /* This should be the last !!! */
+  MDL_TYPE_END};
 
-enum enum_mdl_state { MDL_PENDING, MDL_ACQUIRED };
 
 /** Maximal length of key for metadata locking subsystem. */
 #define MAX_MDLKEY_LENGTH (1 + NAME_LEN + 1 + NAME_LEN + 1)
@@ -77,11 +170,11 @@ public:
     it's necessary to have a separate namespace for them since
     MDL_key is also used outside of the MDL subsystem.
   */
-  enum enum_mdl_namespace { TABLE=0,
+  enum enum_mdl_namespace { GLOBAL=0,
+                            TABLE,
                             FUNCTION,
                             PROCEDURE,
-                            TRIGGER,
-                            GLOBAL };
+                            TRIGGER };
 
   const uchar *ptr() const { return (uchar*) m_ptr; }
   uint length() const { return m_length; }
@@ -217,7 +310,7 @@ public:
     DBUG_ASSERT(ticket == NULL);
     type= type_arg;
   }
-  bool is_shared() const { return type < MDL_INTENTION_EXCLUSIVE; }
+  uint get_deadlock_weight() const;
 
   static MDL_request *create(MDL_key::enum_mdl_namespace mdl_namespace,
                              const char *db, const char *name,
@@ -297,21 +390,25 @@ public:
   void set_cached_object(void *cached_object,
                          mdl_cached_object_release_hook release_hook);
   MDL_context *get_ctx() const { return m_ctx; }
-  bool is_shared() const { return m_type < MDL_INTENTION_EXCLUSIVE; }
   bool is_upgradable_or_exclusive() const
   {
-    return m_type == MDL_SHARED_UPGRADABLE || m_type == MDL_EXCLUSIVE;
+    return m_type == MDL_SHARED_NO_WRITE ||
+           m_type == MDL_SHARED_NO_READ_WRITE ||
+           m_type == MDL_EXCLUSIVE;
   }
-  bool upgrade_shared_lock_to_exclusive();
-  void downgrade_exclusive_lock();
+  enum_mdl_type get_type() const { return m_type; }
+  void downgrade_exclusive_lock(enum_mdl_type type);
+
+  bool has_stronger_or_equal_type(enum_mdl_type type) const;
+
+  bool is_incompatible_when_granted(enum_mdl_type type) const;
+  bool is_incompatible_when_waiting(enum_mdl_type type) const;
+
 private:
   friend class MDL_context;
-  friend class MDL_global_lock;
-  friend class MDL_object_lock;
 
   MDL_ticket(MDL_context *ctx_arg, enum_mdl_type type_arg)
    : m_type(type_arg),
-     m_state(MDL_PENDING),
      m_ctx(ctx_arg),
      m_lock(NULL)
   {}
@@ -321,9 +418,6 @@ private:
 private:
   /** Type of metadata lock. Externally accessible. */
   enum enum_mdl_type m_type;
-  /** State of the metadata lock ticket. Context private. */
-  enum enum_mdl_state m_state;
-
   /**
     Context of the owner of the metadata lock ticket. Externally accessible.
   */
@@ -331,6 +425,7 @@ private:
 
   /** Pointer to the lock object for this lock ticket. Context private. */
   MDL_lock *m_lock;
+
 private:
   MDL_ticket(const MDL_ticket &);               /* not implemented */
   MDL_ticket &operator=(const MDL_ticket &);    /* not implemented */
@@ -359,26 +454,29 @@ public:
 
   typedef Ticket_list::Iterator Ticket_iterator;
 
+  enum mdl_signal_type { NO_WAKE_UP = 0,
+                         NORMAL_WAKE_UP,
+                         VICTIM_WAKE_UP,
+                         TIMEOUT_WAKE_UP };
+
   MDL_context();
   void destroy();
 
-  bool try_acquire_shared_lock(MDL_request *mdl_request);
-  bool acquire_exclusive_lock(MDL_request *mdl_request);
-  bool acquire_exclusive_locks(MDL_request_list *requests);
-  bool try_acquire_exclusive_lock(MDL_request *mdl_request);
+  bool try_acquire_lock(MDL_request *mdl_request);
+  bool acquire_lock(MDL_request *mdl_request);
+  bool acquire_locks(MDL_request_list *requests);
+  bool upgrade_shared_lock_to_exclusive(MDL_ticket *mdl_ticket);
+
   bool clone_ticket(MDL_request *mdl_request);
 
-  bool wait_for_locks(MDL_request_list *requests);
+  bool wait_for_lock(MDL_request *mdl_request);
 
   void release_all_locks_for_name(MDL_ticket *ticket);
   void release_lock(MDL_ticket *ticket);
 
-  bool is_exclusive_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
-                               const char *db,
-                               const char *name);
   bool is_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
-                     const char *db, const char *name);
-
+                     const char *db, const char *name,
+                     enum_mdl_type mdl_type);
 
   bool has_lock(MDL_ticket *mdl_savepoint, MDL_ticket *mdl_ticket);
 
@@ -391,88 +489,208 @@ public:
   {
     /*
       NULL savepoint represents the start of the transaction.
-      Checking for m_lt_or_ha_sentinel also makes sure we never
+      Checking for m_trans_sentinel also makes sure we never
       return a pointer to HANDLER ticket as a savepoint.
     */
-    return m_tickets.front() == m_lt_or_ha_sentinel ? NULL : m_tickets.front();
+    return m_tickets.front() == m_trans_sentinel ? NULL : m_tickets.front();
   }
 
-  void set_lt_or_ha_sentinel()
+  void set_trans_sentinel()
   {
-    m_lt_or_ha_sentinel= mdl_savepoint();
+    m_trans_sentinel= mdl_savepoint();
   }
-  MDL_ticket *lt_or_ha_sentinel() const { return m_lt_or_ha_sentinel; }
+  MDL_ticket *trans_sentinel() const { return m_trans_sentinel; }
 
-  void clear_lt_or_ha_sentinel()
+  void reset_trans_sentinel(MDL_ticket *sentinel_arg)
   {
-    m_lt_or_ha_sentinel= NULL;
+    m_trans_sentinel= sentinel_arg;
   }
-  void move_ticket_after_lt_or_ha_sentinel(MDL_ticket *mdl_ticket);
+  void move_ticket_after_trans_sentinel(MDL_ticket *mdl_ticket);
 
   void release_transactional_locks();
   void rollback_to_savepoint(MDL_ticket *mdl_savepoint);
 
-  bool can_wait_lead_to_deadlock() const;
-
   inline THD *get_thd() const { return m_thd; }
 
+  bool acquire_global_intention_exclusive_lock(MDL_request *mdl_request);
+
   /**
     Wake up context which is waiting for a change of MDL_lock state.
   */
-  void awake()
+  void awake(mdl_signal_type signal)
   {
-    pthread_cond_signal(&m_ctx_wakeup_cond);
+    pthread_mutex_lock(&m_signal_lock);
+    m_signal= signal;
+    pthread_cond_signal(&m_signal_cond);
+    pthread_mutex_unlock(&m_signal_lock);
   }
 
-  bool try_acquire_global_intention_exclusive_lock(MDL_request *mdl_request);
-  bool acquire_global_intention_exclusive_lock(MDL_request *mdl_request);
-
-  bool acquire_global_shared_lock();
-  void release_global_shared_lock();
+  void init(THD *thd_arg) { m_thd= thd_arg; }
 
-  /**
-    Check if this context owns global lock of particular type.
-  */
-  bool is_global_lock_owner(enum_mdl_type type_arg)
+  void set_needs_thr_lock_abort(bool needs_thr_lock_abort)
   {
-    MDL_request mdl_request;
-    bool not_used;
-    mdl_request.init(MDL_key::GLOBAL, "", "", type_arg);
-    return find_ticket(&mdl_request, &not_used);
+    /*
+      @note In theory, this member should be modified under protection
+            of some lock since it can be accessed from different threads.
+            In practice, this is not necessary as code which reads this
+            value and so might miss the fact that value was changed will
+            always re-try reading it after small timeout and therefore
+            will see the new value eventually.
+    */
+    m_needs_thr_lock_abort= TRUE;
+  }
+  bool get_needs_thr_lock_abort() const
+  {
+    return m_needs_thr_lock_abort;
   }
 
-  void init(THD *thd_arg) { m_thd= thd_arg; }
+  bool find_deadlock(Deadlock_detection_context *deadlock_ctx);
 private:
+  /**
+    All MDL tickets acquired by this connection.
+
+    The order of tickets in m_tickets list.
+    ---------------------------------------
+    The entire set of locks acquired by a connection
+    can be separated in two subsets: transactional and
+    non-transactional locks.
+
+    Transactional locks are locks with automatic scope. They
+    are accumulated in the course of a transaction, and
+    released only on COMMIT, ROLLBACK or ROLLBACK TO SAVEPOINT.
+    They must not be (and never are) released manually,
+    i.e. with release_lock() call.
+
+    Non-transactional locks are taken for locks that span
+    multiple transactions or savepoints.
+    These are: HANDLER SQL locks (HANDLER SQL is
+    transaction-agnostic), LOCK TABLES locks (you can COMMIT/etc
+    under LOCK TABLES, and the locked tables stay locked), and
+    SET GLOBAL READ_ONLY=1 global shared lock.
+
+    Transactional locks are always prepended to the beginning
+    of the list. In other words, they are stored in reverse
+    temporal order. Thus, when we rollback to a savepoint,
+    we start popping and releasing tickets from the front
+    until we reach the last ticket acquired after the
+    savepoint.
+
+    Non-transactional locks are always stored after
+    transactional ones, and among each other can be
+    split into three sets:
+
+    [LOCK TABLES locks] [HANDLER locks] [GLOBAL READ LOCK locks]
+
+    The following is known about these sets:
+
+    * we can never have both HANDLER and LOCK TABLES locks
+      together -- HANDLER statements are prohibited under LOCK
+      TABLES, entering LOCK TABLES implicitly closes all open
+      HANDLERs.
+    * GLOBAL READ LOCK locks are always stored after LOCK TABLES
+      locks and after HANDLER locks. This is because one can't say
+      SET GLOBAL read_only=1 or FLUSH TABLES WITH READ LOCK
+      if one has locked tables. One can, however, LOCK TABLES
+      after having entered the read only mode. Note, that
+      subsequent LOCK TABLES statement will unlock the previous
+      set of tables, but not the GRL!
+      There are no HANDLER locks after GRL locks because
+      SET GLOBAL read_only performs a FLUSH TABLES WITH
+      READ LOCK internally, and FLUSH TABLES, in turn, implicitly
+      closes all open HANDLERs.
+      However, one can open a few HANDLERs after entering the
+      read only mode.
+  */
   Ticket_list m_tickets;
   /**
-    This member has two uses:
-    1) When entering LOCK TABLES mode, remember the last taken
-    metadata lock. COMMIT/ROLLBACK must preserve these metadata
-    locks.
-    2) When we have an open HANDLER tables, store the position
-    in the list beyond which we keep locks for HANDLER tables.
-    COMMIT/ROLLBACK must, again, preserve HANDLER metadata locks.
+    Separates transactional and non-transactional locks
+    in m_tickets list, @sa m_tickets.
   */
-  MDL_ticket *m_lt_or_ha_sentinel;
+  MDL_ticket *m_trans_sentinel;
   THD *m_thd;
   /**
+    TRUE -  if for this context we will break protocol and try to
+            acquire table-level locks while having only S lock on
+            some table.
+            To avoid deadlocks which might occur during concurrent
+            upgrade of SNRW lock on such object to X lock we have to
+            abort waits for table-level locks for such connections.
+    FALSE - Otherwise.
+  */
+  bool m_needs_thr_lock_abort;
+
+  /**
+    Read-write lock protecting m_waiting_for member.
+
+    TODO/FIXME: Replace with RW-lock which will prefer readers
+                on all platforms and not only on Linux.
+  */
+  rw_lock_t m_waiting_for_lock;
+  MDL_ticket *m_waiting_for;
+  uint m_deadlock_weight;
+  /**
     Condvar which is used for waiting until this context's pending
     request can be satisfied or this thread has to perform actions
-    to resolve potential deadlock (we subscribe for such notification
-    by adding ticket corresponding to the request to an appropriate
-    queue of waiters).
+    to resolve a potential deadlock (we subscribe to such
+    notification by adding a ticket corresponding to the request
+    to an appropriate queue of waiters).
   */
-  pthread_cond_t m_ctx_wakeup_cond;
+  pthread_mutex_t m_signal_lock;
+  pthread_cond_t m_signal_cond;
+  mdl_signal_type m_signal;
+
 private:
   MDL_ticket *find_ticket(MDL_request *mdl_req,
-                          bool *is_lt_or_ha);
+                          bool *is_transactional);
   void release_locks_stored_before(MDL_ticket *sentinel);
-
-  bool try_acquire_lock_impl(MDL_request *mdl_request);
   bool acquire_lock_impl(MDL_request *mdl_request);
-  bool acquire_exclusive_lock_impl(MDL_request *mdl_request);
 
-  friend bool MDL_ticket::upgrade_shared_lock_to_exclusive();
+  bool find_deadlock();
+
+  void will_wait_for(MDL_ticket *pending_ticket)
+  {
+    rw_wrlock(&m_waiting_for_lock);
+    m_waiting_for= pending_ticket;
+    rw_unlock(&m_waiting_for_lock);
+  }
+
+  void set_deadlock_weight(uint weight)
+  {
+    /*
+      m_deadlock_weight should not be modified while m_waiting_for is
+      non-NULL as in this case this context might participate in deadlock
+      and so m_deadlock_weight can be accessed from other threads.
+    */
+    DBUG_ASSERT(m_waiting_for == NULL);
+    m_deadlock_weight= weight;
+  }
+
+  void stop_waiting()
+  {
+    rw_wrlock(&m_waiting_for_lock);
+    m_waiting_for= NULL;
+    rw_unlock(&m_waiting_for_lock);
+  }
+
+  void wait_reset()
+  {
+    pthread_mutex_lock(&m_signal_lock);
+    m_signal= NO_WAKE_UP;
+    pthread_mutex_unlock(&m_signal_lock);
+  }
+
+  mdl_signal_type wait();
+  mdl_signal_type timed_wait(ulong timeout);
+
+  mdl_signal_type peek_signal()
+  {
+    mdl_signal_type result;
+    pthread_mutex_lock(&m_signal_lock);
+    result= m_signal;
+    pthread_mutex_unlock(&m_signal_lock);
+    return result;
+  }
+
 private:
   MDL_context(const MDL_context &rhs);          /* not implemented */
   MDL_context &operator=(MDL_context &rhs);     /* not implemented */
@@ -487,9 +705,9 @@ void mdl_destroy();
   Functions in the server's kernel used by metadata locking subsystem.
 */
 
-extern bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use);
+extern bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use,
+                                                   bool needs_thr_lock_abort);
 extern void mysql_ha_flush(THD *thd);
-extern void mysql_abort_transactions_with_shared_lock(const MDL_key *mdl_key);
 extern "C" const char *set_thd_proc_info(THD *thd, const char *info,
                                          const char *calling_function,
                                          const char *calling_file,
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 9f153b5aa0e..2316eae4cdb 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1241,8 +1241,9 @@ bool tdc_open_view(THD *thd, TABLE_LIST *table_list, const char *alias,
                    char *cache_key, uint cache_key_length,
                    MEM_ROOT *mem_root, uint flags);
 TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name);
-TABLE *find_write_locked_table(TABLE *list, const char *db,
-                               const char *table_name);
+TABLE *find_table_for_mdl_upgrade(TABLE *list, const char *db,
+                                  const char *table_name,
+                                  bool no_error);
 thr_lock_type read_lock_type_for_table(THD *thd, TABLE *table);
 void execute_init_command(THD *thd, sys_var_str *init_command_var,
 			  rw_lock_t *var_mutex);
@@ -2078,6 +2079,8 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count,
 #define MYSQL_OPEN_SKIP_TEMPORARY               0x0100
 /** Fail instead of waiting when conficting metadata lock is discovered. */
 #define MYSQL_OPEN_FAIL_ON_MDL_CONFLICT         0x0200
+/** Open tables using MDL_SHARED lock instead of one specified in parser. */
+#define MYSQL_OPEN_FORCE_SHARED_MDL             0x0400
 
 /** Please refer to the internals manual. */
 #define MYSQL_OPEN_REOPEN  (MYSQL_LOCK_IGNORE_FLUSH |\
@@ -2098,14 +2101,6 @@ bool mysql_lock_abort_for_thread(THD *thd, TABLE *table);
 MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b);
 TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle,
                                       TABLE_LIST *haystack);
-bool lock_global_read_lock(THD *thd);
-void unlock_global_read_lock(THD *thd);
-bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
-                              bool is_not_commit);
-void start_waiting_global_read_lock(THD *thd);
-bool make_global_read_lock_block_commit(THD *thd);
-bool set_protect_against_global_read_lock(void);
-void unset_protect_against_global_read_lock(void);
 void broadcast_refresh(void);
 
 /* Lock based on name */
diff --git a/sql/set_var.cc b/sql/set_var.cc
index ce7cfcc81a8..0f36a164247 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -4262,7 +4262,7 @@ bool sys_var_opt_readonly::update(THD *thd, set_var *var)
     DBUG_RETURN(true);
   }
 
-  if (thd->global_read_lock)
+  if (thd->global_read_lock.is_acquired())
   {
     /*
       This connection already holds the global read lock.
@@ -4285,8 +4285,8 @@ bool sys_var_opt_readonly::update(THD *thd, set_var *var)
     [3] prevents transactions from being committed.
   */
 
-  if (lock_global_read_lock(thd))
-    DBUG_RETURN(true);
+  if (thd->global_read_lock.lock_global_read_lock(thd))
+    DBUG_RETURN(TRUE);
 
   /*
     This call will be blocked by any connection holding a READ or WRITE lock.
@@ -4300,7 +4300,7 @@ bool sys_var_opt_readonly::update(THD *thd, set_var *var)
   if ((result= close_cached_tables(thd, NULL, FALSE, TRUE)))
     goto end_with_read_lock;
 
-  if ((result= make_global_read_lock_block_commit(thd)))
+  if ((result= thd->global_read_lock.make_global_read_lock_block_commit(thd)))
     goto end_with_read_lock;
 
   /* Change the opt_readonly system variable, safe because the lock is held */
@@ -4308,7 +4308,7 @@ bool sys_var_opt_readonly::update(THD *thd, set_var *var)
 
 end_with_read_lock:
   /* Release the lock */
-  unlock_global_read_lock(thd);
+  thd->global_read_lock.unlock_global_read_lock(thd);
   DBUG_RETURN(result);
 }
 
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index cd7486dec72..83c928eb88f 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -3966,7 +3966,8 @@ sp_head::add_used_tables_to_table_list(THD *thd,
       table->belong_to_view= belong_to_view;
       table->trg_event_map= stab->trg_event_map;
       table->mdl_request.init(MDL_key::TABLE, table->db, table->table_name,
-                              MDL_SHARED);
+                              table->lock_type >= TL_WRITE_ALLOW_WRITE ?
+                              MDL_SHARED_WRITE : MDL_SHARED_READ);
 
       /* Everyting else should be zeroed */
 
@@ -4009,7 +4010,8 @@ sp_add_to_query_tables(THD *thd, LEX *lex,
   table->select_lex= lex->current_select;
   table->cacheable_table= 1;
   table->mdl_request.init(MDL_key::TABLE, table->db, table->table_name,
-                          MDL_SHARED);
+                          table->lock_type >= TL_WRITE_ALLOW_WRITE ?
+                          MDL_SHARED_WRITE : MDL_SHARED_READ);
 
   lex->add_to_query_tables(table);
   return table;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 2f891375163..8dced67dd44 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -131,8 +131,6 @@ static bool tdc_wait_for_old_versions(THD *thd,
 static bool
 has_write_table_with_auto_increment(TABLE_LIST *tables);
 
-TABLE *find_table_for_mdl_upgrade(TABLE *list, const char *db,
-                                  const char *table_name);
 
 uint cached_open_tables(void)
 {
@@ -477,8 +475,10 @@ TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
     To be able perform any operation on table we should own
     some kind of metadata lock on it.
   */
-  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, table_list->db,
-                                             table_list->table_name));
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE,
+                                             table_list->db,
+                                             table_list->table_name,
+                                             MDL_SHARED));
 
   /* Read table definition from cache */
   if ((share= (TABLE_SHARE*) my_hash_search(&table_def_cache,(uchar*) key,
@@ -1003,7 +1003,7 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
     {
       /* A check that the table was locked for write is done by the caller. */
       TABLE *table= find_table_for_mdl_upgrade(thd->open_tables, table_list->db,
-                                               table_list->table_name);
+                                               table_list->table_name, TRUE);
 
       /* May return NULL if this table has already been closed via an alias. */
       if (! table)
@@ -1084,7 +1084,7 @@ err_with_reopen:
       than picking only those tables that were flushed.
     */
     for (TABLE *tab= thd->open_tables; tab; tab= tab->next)
-      tab->mdl_ticket->downgrade_exclusive_lock();
+      tab->mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
   }
   DBUG_RETURN(result);
 }
@@ -2145,7 +2145,7 @@ bool wait_while_table_is_used(THD *thd, TABLE *table,
                        table->s->table_name.str, (ulong) table->s,
                        table->db_stat, table->s->version));
 
-  if (table->mdl_ticket->upgrade_shared_lock_to_exclusive())
+  if (thd->mdl_context.upgrade_shared_lock_to_exclusive(table->mdl_ticket))
     DBUG_RETURN(TRUE);
 
   pthread_mutex_lock(&LOCK_open);
@@ -2336,6 +2336,7 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
 {
   if (table_list->lock_strategy)
   {
+    MDL_request_list mdl_requests;
     MDL_request *global_request;
     /*
       In case of CREATE TABLE .. If NOT EXISTS .. SELECT, the table
@@ -2350,67 +2351,51 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
     mdl_request->set_type(MDL_EXCLUSIVE);
     DBUG_ASSERT(! thd->mdl_context.has_locks() ||
                 thd->handler_tables_hash.records ||
-                thd->global_read_lock);
+                thd->global_read_lock.is_acquired());
 
     if (!(global_request= ot_ctx->get_global_mdl_request(thd)))
       return 1;
 
-    if (! global_request->ticket)
-    {
-      ot_ctx->add_request(global_request);
-      if (thd->mdl_context.acquire_global_intention_exclusive_lock(
-                             global_request))
-      return 1;
-    }
+    mdl_requests.push_front(mdl_request);
+    mdl_requests.push_front(global_request);
 
-    ot_ctx->add_request(mdl_request);
-    if (thd->mdl_context.acquire_exclusive_lock(mdl_request))
+    if (thd->mdl_context.acquire_locks(&mdl_requests))
       return 1;
   }
   else
   {
-    /*
-      There is no MDL_SHARED_UPGRADABLE_HIGH_PRIO type of metadata lock so we
-      want to be sure that caller doesn't pass us both flags simultaneously.
-    */
-    DBUG_ASSERT(!(flags & MYSQL_OPEN_TAKE_UPGRADABLE_MDL) ||
-                !(flags & MYSQL_LOCK_IGNORE_FLUSH));
-
-    if (flags & MYSQL_OPEN_TAKE_UPGRADABLE_MDL &&
-        table_list->lock_type >= TL_WRITE_ALLOW_WRITE)
-      mdl_request->set_type(MDL_SHARED_UPGRADABLE);
-    if (flags & MYSQL_LOCK_IGNORE_FLUSH)
-      mdl_request->set_type(MDL_SHARED_HIGH_PRIO);
-
-    if (mdl_request->type == MDL_SHARED_UPGRADABLE)
+    if (flags & MYSQL_OPEN_FORCE_SHARED_MDL)
     {
-      MDL_request *global_request;
+      /*
+        While executing PREPARE for prepared statement we override
+        type-of-operation aware type of shared metadata lock which
+        was set in the parser with simple shared metadata lock.
+        This is necessary to allow concurrent execution of PREPARE
+        and LOCK TABLES WRITE statement which locks one of the tables
+        used in the statement being prepared.
+      */
+      DBUG_ASSERT(!(flags & (MYSQL_OPEN_TAKE_UPGRADABLE_MDL |
+                             MYSQL_LOCK_IGNORE_FLUSH)));
 
-      if (!(global_request= ot_ctx->get_global_mdl_request(thd)))
-        return 1;
-      if (! global_request->ticket)
-      {
-        ot_ctx->add_request(global_request);
-        if (thd->mdl_context.try_acquire_global_intention_exclusive_lock(
-                               global_request))
-          return 1;
-        if (! global_request->ticket)
-          goto failure;
-      }
+      mdl_request->set_type(MDL_SHARED);
+    }
+    else if (flags & MYSQL_LOCK_IGNORE_FLUSH)
+    {
+      DBUG_ASSERT(!(flags & MYSQL_OPEN_TAKE_UPGRADABLE_MDL));
+      mdl_request->set_type(MDL_SHARED_HIGH_PRIO);
     }
 
     ot_ctx->add_request(mdl_request);
 
-    if (thd->mdl_context.try_acquire_shared_lock(mdl_request))
+    if (thd->mdl_context.try_acquire_lock(mdl_request))
       return 1;
 
-failure:
     if (mdl_request->ticket == NULL)
     {
       if (flags & MYSQL_OPEN_FAIL_ON_MDL_CONFLICT)
         my_error(ER_WARN_I_S_SKIPPED_TABLE, MYF(0), table_list->db, table_list->table_name);
       else
-        (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT);
+        ot_ctx->request_backoff_action(Open_table_context::OT_WAIT_MDL_LOCK);
       return 1;
     }
   }
@@ -2563,6 +2548,17 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
         {
           int distance= ((int) table->reginfo.lock_type -
                          (int) table_list->lock_type);
+
+          /*
+            If we are performing DDL operation we also should ensure
+            that we will find TABLE instance with upgradable metadata
+            lock,
+          */
+          if ((flags & MYSQL_OPEN_TAKE_UPGRADABLE_MDL) &&
+              table_list->lock_type >= TL_WRITE_ALLOW_WRITE &&
+              ! table->mdl_ticket->is_upgradable_or_exclusive())
+            distance= -1;
+
           /*
             Find a table that either has the exact lock type requested,
             or has the best suitable lock. In case there is no locked
@@ -2596,6 +2592,13 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     }
     if (best_table)
     {
+      if ((flags & MYSQL_OPEN_TAKE_UPGRADABLE_MDL) &&
+          table_list->lock_type >= TL_WRITE_ALLOW_WRITE &&
+          ! best_table->mdl_ticket->is_upgradable_or_exclusive())
+      {
+        my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), alias);
+        DBUG_RETURN(TRUE);
+      }
       table= best_table;
       table->query_id= thd->query_id;
       DBUG_PRINT("info",("Using locked table"));
@@ -2610,8 +2613,10 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
       TABLES breaks metadata locking protocol (potentially can lead
       to deadlocks) it should be disallowed.
     */
-    if (thd->mdl_context.is_lock_owner(MDL_key::TABLE, table_list->db,
-                                       table_list->table_name))
+    if (thd->mdl_context.is_lock_owner(MDL_key::TABLE,
+                                       table_list->db,
+                                       table_list->table_name,
+                                       MDL_SHARED))
     {
       char path[FN_REFLEN + 1];
       enum legacy_db_type not_used;
@@ -2688,7 +2693,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   {
     /* Someone did a refresh while thread was opening tables */
     pthread_mutex_unlock(&LOCK_open);
-    (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT);
+    (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT_TDC);
     DBUG_RETURN(TRUE);
   }
 
@@ -2828,7 +2833,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
        */
       release_table_share(share);
       pthread_mutex_unlock(&LOCK_open);
-      (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT);
+      (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT_TDC);
       DBUG_RETURN(TRUE);
     }
     /* Force close at once after usage */
@@ -2895,11 +2900,11 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   /*
     In CREATE TABLE .. If NOT EXISTS .. SELECT we have found that
     table exists now we should downgrade our exclusive metadata
-    lock on this table to shared metadata lock.
+    lock on this table to SW metadata lock.
   */
   if (table_list->lock_strategy == TABLE_LIST::EXCLUSIVE_DOWNGRADABLE_MDL &&
       !(flags & MYSQL_OPEN_HAS_MDL_LOCK))
-    mdl_ticket->downgrade_exclusive_lock();
+    mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_WRITE);
 
   table->mdl_ticket= mdl_ticket;
 
@@ -2982,33 +2987,43 @@ TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name)
 
 
 /**
-   Find write locked instance of table in the list of open tables,
-   emit error if no such instance found.
+   Find instance of TABLE with upgradable or exclusive metadata
+   lock from the list of open tables, emit error if no such table
+   found.
 
-   @param thd        List of TABLE objects to be searched
+   @param list       List of TABLE objects to be searched
    @param db         Database name.
    @param table_name Name of table.
+   @param no_error   Don't emit error if no suitable TABLE
+                     instance were found.
 
-   @return Pointer to write-locked TABLE instance, 0 - otherwise.
+   @return Pointer to TABLE instance with MDL_SHARED_NO_WRITE,
+           MDL_SHARED_NO_READ_WRITE, or MDL_EXCLUSIVE metadata
+           lock, NULL otherwise.
 */
 
-TABLE *find_write_locked_table(TABLE *list, const char *db, const char *table_name)
+TABLE *find_table_for_mdl_upgrade(TABLE *list, const char *db,
+                                  const char *table_name,
+                                  bool no_error)
 {
   TABLE *tab= find_locked_table(list, db, table_name);
 
   if (!tab)
   {
-    my_error(ER_TABLE_NOT_LOCKED, MYF(0), table_name);
-    return 0;
+    if (!no_error)
+      my_error(ER_TABLE_NOT_LOCKED, MYF(0), table_name);
+    return NULL;
   }
   else
   {
-    while (tab->reginfo.lock_type < TL_WRITE_LOW_PRIORITY &&
+    while (tab->mdl_ticket != NULL &&
+           !tab->mdl_ticket->is_upgradable_or_exclusive() &&
            (tab= find_locked_table(tab->next, db, table_name)))
       continue;
     if (!tab)
     {
-      my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_name);
+      if (!no_error)
+        my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_name);
       return 0;
     }
   }
@@ -3016,34 +3031,6 @@ TABLE *find_write_locked_table(TABLE *list, const char *db, const char *table_na
 }
 
 
-/**
-   Find instance of TABLE with MDL_SHARED_UPGRADABLE or
-   MDL_EXCLUSIVE lock from the list of open tables.
-
-   @param list       List of TABLE objects to be searched
-   @param db         Database name.
-   @param table_name Name of table.
-
-   @return Pointer to MDL_SHARED_UPGRADABLE or MDL_EXCLUSIVE
-           TABLE instance, NULL otherwise.
-*/
-
-TABLE *find_table_for_mdl_upgrade(TABLE *list, const char *db,
-                                  const char *table_name)
-{
-  TABLE *tab= find_locked_table(list, db, table_name);
-
-  while (tab != NULL)
-  {
-    if (tab->mdl_ticket != NULL &&
-        tab->mdl_ticket->is_upgradable_or_exclusive())
-      return tab;
-    tab= find_locked_table(tab->next, db, table_name);
-  }
-  return NULL;
-}
-
-
 /***********************************************************************
   class Locked_tables_list implementation. Declared in sql_class.h
 ************************************************************************/
@@ -3741,9 +3728,9 @@ end_with_lock_open:
 Open_table_context::Open_table_context(THD *thd)
   :m_action(OT_NO_ACTION),
    m_start_of_statement_svp(thd->mdl_context.mdl_savepoint()),
-   m_has_locks((thd->in_multi_stmt_transaction() ||
-                thd->mdl_context.lt_or_ha_sentinel()) &&
-               thd->mdl_context.has_locks()),
+   m_has_locks((thd->in_multi_stmt_transaction() &&
+                thd->mdl_context.has_locks()) ||
+                thd->mdl_context.trans_sentinel()),
    m_global_mdl_request(NULL)
 {}
 
@@ -3798,7 +3785,7 @@ request_backoff_action(enum_open_table_action action_arg)
     Waiting will be done after releasing metadata locks acquired
     by this statement.
   */
-  if (m_has_locks && action_arg != OT_WAIT)
+  if (m_has_locks && action_arg != OT_WAIT_MDL_LOCK)
   {
     my_error(ER_LOCK_DEADLOCK, MYF(0));
     return TRUE;
@@ -3838,33 +3825,28 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
   /* Execute the action. */
   switch (m_action)
   {
-    case OT_WAIT:
-      result= (thd->mdl_context.wait_for_locks(&m_mdl_requests) ||
-               tdc_wait_for_old_versions(thd, &m_mdl_requests));
+    case OT_WAIT_MDL_LOCK:
+      result= thd->mdl_context.wait_for_lock(mdl_request);
+      break;
+    case OT_WAIT_TDC:
+      result= tdc_wait_for_old_versions(thd, &m_mdl_requests);
       DBUG_ASSERT(thd->mysys_var->current_mutex == NULL);
       break;
     case OT_DISCOVER:
       {
         MDL_request mdl_global_request;
         MDL_request mdl_xlock_request(mdl_request);
+        MDL_request_list mdl_requests;
 
         mdl_global_request.init(MDL_key::GLOBAL, "", "",
                                 MDL_INTENTION_EXCLUSIVE);
         mdl_xlock_request.set_type(MDL_EXCLUSIVE);
 
+        mdl_requests.push_front(&mdl_xlock_request);
+        mdl_requests.push_front(&mdl_global_request);
 
-        if ((result= thd->mdl_context.acquire_global_intention_exclusive_lock(
-                                        &mdl_global_request)))
-          break;
-
-        if ((result=
-             thd->mdl_context.acquire_exclusive_lock(&mdl_xlock_request)))
-        {
-          /*
-            We rely on close_thread_tables() to release global lock eventually.
-          */
+        if ((result= thd->mdl_context.acquire_locks(&mdl_requests)))
           break;
-        }
 
         DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::TABLE);
         pthread_mutex_lock(&LOCK_open);
@@ -3885,23 +3867,17 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
       {
         MDL_request mdl_global_request;
         MDL_request mdl_xlock_request(mdl_request);
+        MDL_request_list mdl_requests;
 
         mdl_global_request.init(MDL_key::GLOBAL, "", "",
                                 MDL_INTENTION_EXCLUSIVE);
         mdl_xlock_request.set_type(MDL_EXCLUSIVE);
 
-        if ((result= thd->mdl_context.acquire_global_intention_exclusive_lock(
-                                        &mdl_global_request)))
-          break;
+        mdl_requests.push_front(&mdl_xlock_request);
+        mdl_requests.push_front(&mdl_global_request);
 
-        if ((result=
-             thd->mdl_context.acquire_exclusive_lock(&mdl_xlock_request)))
-        {
-          /*
-            We rely on close_thread_tables() to release global lock eventually.
-          */
+        if ((result= thd->mdl_context.acquire_locks(&mdl_requests)))
           break;
-        }
 
         DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::TABLE);
         pthread_mutex_lock(&LOCK_open);
@@ -4015,13 +3991,13 @@ open_and_process_routine(THD *thd, Query_tables_list *prelocking_ctx,
         */
         DBUG_ASSERT(rt->mdl_request.type == MDL_SHARED);
 
-        if (thd->mdl_context.try_acquire_shared_lock(&rt->mdl_request))
+        if (thd->mdl_context.try_acquire_lock(&rt->mdl_request))
           DBUG_RETURN(TRUE);
 
         if (rt->mdl_request.ticket == NULL)
         {
           /* A lock conflict. Someone's trying to modify SP metadata. */
-          ot_ctx->request_backoff_action(Open_table_context::OT_WAIT);
+          ot_ctx->request_backoff_action(Open_table_context::OT_WAIT_MDL_LOCK);
           DBUG_RETURN(TRUE);
         }
         DEBUG_SYNC(thd, "after_shared_lock_pname");
@@ -4345,6 +4321,66 @@ end:
 
 
 /**
+  Acquire upgradable (SNW, SNRW) metadata locks on tables to be opened
+  for LOCK TABLES or a DDL statement.
+
+  @param thd           Thread context.
+  @param tables_start  Start of list of tables on which upgradable locks
+                       should be acquired.
+  @param tables_end    End of list of tables.
+  @param ot_ctx        Context of open_tables() operation.
+
+  @retval FALSE  Success.
+  @retval TRUE   Failure (e.g. connection was killed)
+*/
+
+static bool
+open_tables_acquire_upgradable_mdl(THD *thd, TABLE_LIST *tables_start,
+                                   TABLE_LIST *tables_end,
+                                   Open_table_context *ot_ctx)
+{
+  MDL_request_list mdl_requests;
+  TABLE_LIST *table;
+
+  for (table= tables_start; table && table != tables_end;
+       table= table->next_global)
+  {
+    if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
+    {
+      table->mdl_request.set_type(table->lock_type > TL_WRITE_ALLOW_READ ?
+                                  MDL_SHARED_NO_READ_WRITE :
+                                  MDL_SHARED_NO_WRITE);
+      mdl_requests.push_front(&table->mdl_request);
+    }
+  }
+
+  if (! mdl_requests.is_empty())
+  {
+    MDL_request *global_request= ot_ctx->get_global_mdl_request(thd);
+
+    if (global_request == NULL)
+      return TRUE;
+    mdl_requests.push_front(global_request);
+  }
+
+  if (thd->mdl_context.acquire_locks(&mdl_requests))
+    return TRUE;
+
+  for (table= tables_start; table && table != tables_end;
+       table= table->next_global)
+  {
+    if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
+    {
+      table->mdl_request.ticket= NULL;
+      table->mdl_request.set_type(MDL_SHARED_WRITE);
+    }
+  }
+
+  return FALSE;
+}
+
+
+/**
   Open all tables in list
 
   @param[in]     thd      Thread context.
@@ -4419,6 +4455,29 @@ restart:
   thd_proc_info(thd, "Opening tables");
 
   /*
+    If we are executing LOCK TABLES statement or a DDL statement
+    (in non-LOCK TABLES mode) we might have to acquire upgradable
+    semi-exclusive metadata locks (SNW or SNRW) on some of the
+    tables to be opened.
+    So we acquire all such locks at once here as doing this in one
+    by one fashion may lead to deadlocks or starvation. Later when
+    we will be opening corresponding table pre-acquired metadata
+    lock will be reused (thanks to the fact that in recursive case
+    metadata locks are acquired without waiting).
+  */
+  if ((flags & MYSQL_OPEN_TAKE_UPGRADABLE_MDL) &&
+      ! thd->locked_tables_mode)
+  {
+    if (open_tables_acquire_upgradable_mdl(thd, *start,
+                                           thd->lex->first_not_own_table(),
+                                           &ot_ctx))
+    {
+      error= TRUE;
+      goto err;
+    }
+  }
+
+  /*
     Perform steps of prelocking algorithm until there are unprocessed
     elements in prelocking list/set.
   */
@@ -4922,8 +4981,8 @@ retry:
   while ((error= open_table(thd, table_list, thd->mem_root, &ot_ctx, 0)) &&
          ot_ctx.can_recover_from_failed_open())
   {
-    /* We never have an open HANDLER or LOCK TABLES here. */
-    DBUG_ASSERT(thd->mdl_context.lt_or_ha_sentinel() == NULL);
+    /* We never have an open HANDLER, LOCK TABLES or GRL here. */
+    DBUG_ASSERT(thd->mdl_context.trans_sentinel() == NULL);
     /*
       Even though we have failed to open table we still need to
       call release_transactional_locks() to release metadata locks which
@@ -4974,8 +5033,8 @@ retry:
             close_thread_tables(thd);
             table_list->table= NULL;
             table_list->mdl_request.ticket= NULL;
-            /* We never have an open HANDLER or LOCK TABLES here. */
-            DBUG_ASSERT(thd->mdl_context.lt_or_ha_sentinel() == NULL);
+            /* We never have an open HANDLER, LOCK TABLES or GRL here. */
+            DBUG_ASSERT(thd->mdl_context.trans_sentinel() == NULL);
             thd->mdl_context.rollback_to_savepoint(ot_ctx.start_of_statement_svp());
             goto retry;
           }
@@ -8459,15 +8518,19 @@ void flush_tables()
 
    @param thd    Current thread context
    @param in_use The thread to wake up
+   @param needs_thr_lock_abort Indicates that to wake up thread
+                               this call needs to abort its waiting
+                               on table-level lock.
 
    @retval  TRUE  if the thread was woken up
-   @retval  FALSE otherwise (e.g. it was not waiting for a table-level lock).
+   @retval  FALSE otherwise.
 
    @note It is one of two places where border between MDL and the
          rest of the server is broken.
 */
 
-bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use)
+bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use,
+                                            bool needs_thr_lock_abort)
 {
   bool signalled= FALSE;
   if ((in_use->system_thread & SYSTEM_THREAD_DELAYED_INSERT) &&
@@ -8481,19 +8544,23 @@ bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use)
     signalled= TRUE;
   }
   pthread_mutex_lock(&LOCK_open);
-  for (TABLE *thd_table= in_use->open_tables;
-       thd_table ;
-       thd_table= thd_table->next)
+
+  if (needs_thr_lock_abort)
   {
-    /*
-      Check for TABLE::needs_reopen() is needed since in some places we call
-      handler::close() for table instance (and set TABLE::db_stat to 0)
-      and do not remove such instances from the THD::open_tables
-      for some time, during which other thread can see those instances
-      (e.g. see partitioning code).
-    */
-    if (!thd_table->needs_reopen())
-      signalled|= mysql_lock_abort_for_thread(thd, thd_table);
+    for (TABLE *thd_table= in_use->open_tables;
+         thd_table ;
+         thd_table= thd_table->next)
+    {
+      /*
+        Check for TABLE::needs_reopen() is needed since in some places we call
+        handler::close() for table instance (and set TABLE::db_stat to 0)
+        and do not remove such instances from the THD::open_tables
+        for some time, during which other thread can see those instances
+        (e.g. see partitioning code).
+        */
+      if (!thd_table->needs_reopen())
+        signalled|= mysql_lock_abort_for_thread(thd, thd_table);
+    }
   }
   /*
     Wake up threads waiting in tdc_wait_for_old_versions().
@@ -8512,28 +8579,6 @@ bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use)
 
 
 /**
-  Force transactions holding shared metadata lock on the table to call
-  MDL_context::can_wait_lead_to_deadlock() even if they don't need any
-  new metadata locks so they can detect potential deadlocks between
-  metadata locking subsystem and table-level locks.
-
-  @param mdl_key MDL key for the table on which we are upgrading
-                 metadata lock.
-*/
-
-void mysql_abort_transactions_with_shared_lock(const MDL_key *mdl_key)
-{
-  if (mdl_key->mdl_namespace() == MDL_key::TABLE)
-  {
-    pthread_mutex_lock(&LOCK_open);
-    tdc_remove_table(NULL, TDC_RT_REMOVE_UNUSED, mdl_key->db_name(),
-                     mdl_key->name());
-    pthread_mutex_unlock(&LOCK_open);
-  }
-}
-
-
-/**
    Remove all or some (depending on parameter) instances of TABLE and
    TABLE_SHARE from the table definition cache.
 
@@ -8574,8 +8619,8 @@ void tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
   safe_mutex_assert_owner(&LOCK_open);
 
   DBUG_ASSERT(remove_type == TDC_RT_REMOVE_UNUSED ||
-              thd->mdl_context.is_exclusive_lock_owner(MDL_key::TABLE,
-                                                       db, table_name));
+              thd->mdl_context.is_lock_owner(MDL_key::TABLE, db, table_name,
+                                             MDL_EXCLUSIVE));
 
   key_length=(uint) (strmov(strmov(key,db)+1,table_name)-key)+1;
 
@@ -8639,24 +8684,6 @@ tdc_wait_for_old_versions(THD *thd, MDL_request_list *mdl_requests)
     */
     mysql_ha_flush(thd);
 
-    /*
-      Check if there is someone waiting for one of metadata locks
-      held by this connection and return an error if that's the
-      case, since this situation may lead to a deadlock.
-      This can happen, when, for example, this connection is
-      waiting for an old version of some table to go away and
-      another connection is trying to upgrade its shared
-      metadata lock to exclusive, and thus is waiting
-      for this to release its lock. We must check for
-      the condition on each iteration of the loop to remove
-      any window for a race.
-    */
-    if (thd->mdl_context.can_wait_lead_to_deadlock())
-    {
-      my_error(ER_LOCK_DEADLOCK, MYF(0));
-      return TRUE;
-    }
-
     pthread_mutex_lock(&LOCK_open);
 
     MDL_request_list::Iterator it(*mdl_requests);
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 62de06d382c..8affe88d01d 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -452,8 +452,6 @@ THD::THD()
    examined_row_count(0),
    warning_info(&main_warning_info),
    stmt_da(&main_da),
-   global_read_lock_protection(0),
-   global_read_lock(0),
    is_fatal_error(0),
    transaction_rollback_request(0),
    is_fatal_sub_stmt_error(0),
@@ -1000,22 +998,22 @@ void THD::cleanup(void)
   locked_tables_list.unlock_locked_tables(this);
   mysql_ha_cleanup(this);
 
+  DBUG_ASSERT(open_tables == NULL);
   /*
     If the thread was in the middle of an ongoing transaction (rolled
     back a few lines above) or under LOCK TABLES (unlocked the tables
     and left the mode a few lines above), there will be outstanding
     metadata locks. Release them.
   */
-  DBUG_ASSERT(open_tables == NULL);
-  /* All HANDLERs must have been closed by now. */
-  DBUG_ASSERT(mdl_context.lt_or_ha_sentinel() == NULL ||
-              global_read_lock);
-  /*
-    Due to the above assert, this is guaranteed to release *all* in
-    this session.
-  */
   mdl_context.release_transactional_locks();
 
+  /* Release the global read lock, if acquired. */
+  if (global_read_lock.is_acquired())
+    global_read_lock.unlock_global_read_lock(this);
+
+  /* All metadata locks must have been released by now. */
+  DBUG_ASSERT(!mdl_context.has_locks());
+
 #if defined(ENABLED_DEBUG_SYNC)
   /* End the Debug Sync Facility. See debug_sync.cc. */
   debug_sync_end_thread(this);
@@ -1031,8 +1029,6 @@ void THD::cleanup(void)
   sp_cache_clear(&sp_proc_cache);
   sp_cache_clear(&sp_func_cache);
 
-  if (global_read_lock)
-    unlock_global_read_lock(this);
   if (ull)
   {
     pthread_mutex_lock(&LOCK_user_locks);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index dce06f7e0c5..2ca67133fe5 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1287,7 +1287,8 @@ public:
   enum enum_open_table_action
   {
     OT_NO_ACTION= 0,
-    OT_WAIT,
+    OT_WAIT_MDL_LOCK,
+    OT_WAIT_TDC,
     OT_DISCOVER,
     OT_REPAIR
   };
@@ -1427,6 +1428,45 @@ struct Ha_data
   Ha_data() :ha_ptr(NULL) {}
 };
 
+/**
+  An instance of the global read lock in a connection.
+  Implemented in lock.cc.
+*/
+
+class Global_read_lock
+{
+public:
+  enum enum_grl_state
+  {
+    GRL_NONE,
+    GRL_ACQUIRED,
+    GRL_ACQUIRED_AND_BLOCKS_COMMIT
+  };
+
+  Global_read_lock()
+    :m_protection_count(0), m_state(GRL_NONE), m_mdl_global_shared_lock(NULL)
+  {}
+
+  bool lock_global_read_lock(THD *thd);
+  void unlock_global_read_lock(THD *thd);
+  bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
+                                bool is_not_commit);
+  void start_waiting_global_read_lock(THD *thd);
+  bool make_global_read_lock_block_commit(THD *thd);
+  bool is_acquired() const { return m_state != GRL_NONE; }
+  bool has_protection() const { return m_protection_count > 0; }
+  MDL_ticket *global_shared_lock() const { return m_mdl_global_shared_lock; }
+private:
+  uint           m_protection_count;            // GRL protection count
+  /**
+    In order to acquire the global read lock, the connection must
+    acquire a global shared metadata lock, to prohibit all DDL.
+  */
+  enum_grl_state m_state;
+  MDL_ticket *m_mdl_global_shared_lock;
+};
+
+
 extern "C" void my_message_sql(uint error, const char *str, myf MyFlags);
 
 /**
@@ -1685,6 +1725,7 @@ public:
       init_sql_alloc(&mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);
     }
   } transaction;
+  Global_read_lock global_read_lock;
   Field      *dup_field;
 #ifndef __WIN__
   sigset_t signals;
@@ -1905,8 +1946,7 @@ public:
   ulong	     rand_saved_seed1, rand_saved_seed2;
   pthread_t  real_id;                           /* For debugging */
   my_thread_id  thread_id;
-  uint       global_read_lock_protection;// GRL protection count
-  uint	     tmp_table, global_read_lock;
+  uint	     tmp_table;
   uint	     server_status,open_options;
   enum enum_thread_type system_thread;
   uint       select_number;             //number of select (used for EXPLAIN)
@@ -2585,15 +2625,16 @@ public:
   void enter_locked_tables_mode(enum_locked_tables_mode mode_arg)
   {
     DBUG_ASSERT(locked_tables_mode == LTM_NONE);
-    DBUG_ASSERT(! mdl_context.lt_or_ha_sentinel() ||
-                mdl_context.is_global_lock_owner(MDL_SHARED));
-    mdl_context.set_lt_or_ha_sentinel();
+    DBUG_ASSERT(handler_tables_hash.records == 0);
+
+    mdl_context.set_trans_sentinel();
     locked_tables_mode= mode_arg;
   }
   void leave_locked_tables_mode()
   {
     locked_tables_mode= LTM_NONE;
-    mdl_context.clear_lt_or_ha_sentinel();
+    /* Make sure we don't release the global read lock when leaving LTM. */
+    mdl_context.reset_trans_sentinel(global_read_lock.global_shared_lock());
   }
 private:
   /** The current internal error handler for this thread, or NULL. */
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 1b61a96d7ff..e7a36f664dd 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -637,7 +637,7 @@ int mysql_create_db(THD *thd, char *db, HA_CREATE_INFO *create_info,
     has the global read lock and refuses the operation with
     ER_CANT_UPDATE_WITH_READLOCK if applicable.
   */
-  if (wait_if_global_read_lock(thd, 0, 1))
+  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
   {
     error= -1;
     goto exit2;
@@ -754,7 +754,7 @@ not_silent:
 
 exit:
   pthread_mutex_unlock(&LOCK_mysql_create_db);
-  start_waiting_global_read_lock(thd);
+  thd->global_read_lock.start_waiting_global_read_lock(thd);
 exit2:
   DBUG_RETURN(error);
 }
@@ -781,7 +781,7 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
     has the global read lock and refuses the operation with
     ER_CANT_UPDATE_WITH_READLOCK if applicable.
   */
-  if ((error=wait_if_global_read_lock(thd,0,1)))
+  if ((error= thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE)))
     goto exit2;
 
   pthread_mutex_lock(&LOCK_mysql_create_db);
@@ -831,7 +831,7 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
 
 exit:
   pthread_mutex_unlock(&LOCK_mysql_create_db);
-  start_waiting_global_read_lock(thd);
+  thd->global_read_lock.start_waiting_global_read_lock(thd);
 exit2:
   DBUG_RETURN(error);
 }
@@ -876,7 +876,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     has the global read lock and refuses the operation with
     ER_CANT_UPDATE_WITH_READLOCK if applicable.
   */
-  if (wait_if_global_read_lock(thd, 0, 1))
+  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
   {
     error= -1;
     goto exit2;
@@ -1027,7 +1027,7 @@ exit:
   if (thd->db && !strcmp(thd->db, db) && error == 0)
     mysql_change_db_impl(thd, NULL, 0, thd->variables.collation_server);
   pthread_mutex_unlock(&LOCK_mysql_create_db);
-  start_waiting_global_read_lock(thd);
+  thd->global_read_lock.start_waiting_global_read_lock(thd);
 exit2:
   DBUG_RETURN(error);
 }
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 228c001f71b..87e59c0ac65 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -1100,7 +1100,6 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   TABLE *table;
   bool error= TRUE;
   uint path_length;
-  MDL_request mdl_global_request, mdl_request;
   /*
     Is set if we're under LOCK TABLES, and used
     to downgrade the exclusive lock after the
@@ -1186,8 +1185,8 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
 
     if (thd->locked_tables_mode)
     {
-      if (!(table= find_write_locked_table(thd->open_tables, table_list->db,
-                                           table_list->table_name)))
+      if (!(table= find_table_for_mdl_upgrade(thd->open_tables, table_list->db,
+                                              table_list->table_name, FALSE)))
         DBUG_RETURN(TRUE);
       mdl_ticket= table->mdl_ticket;
       if (wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN))
@@ -1196,6 +1195,8 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
     }
     else
     {
+      MDL_request mdl_global_request, mdl_request;
+      MDL_request_list mdl_requests;
       /*
         Even though we could use the previous execution branch
         here just as well, we must not try to open the table: 
@@ -1211,17 +1212,12 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
       mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
       mdl_request.init(MDL_key::TABLE, table_list->db, table_list->table_name,
                        MDL_EXCLUSIVE);
-      if (thd->mdl_context.acquire_global_intention_exclusive_lock(
-                             &mdl_global_request))
-        DBUG_RETURN(TRUE);
-      if (thd->mdl_context.acquire_exclusive_lock(&mdl_request))
-      {
-        /*
-          We rely on that close_thread_tables() to release global lock
-          in this case.
-        */
+      mdl_requests.push_front(&mdl_request);
+      mdl_requests.push_front(&mdl_global_request);
+
+      if (thd->mdl_context.acquire_locks(&mdl_requests))
         DBUG_RETURN(TRUE);
-      }
+
       has_mdl_lock= TRUE;
       pthread_mutex_lock(&LOCK_open);
       tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_list->db,
@@ -1263,7 +1259,7 @@ end:
     if (has_mdl_lock)
       thd->mdl_context.release_transactional_locks();
     if (mdl_ticket)
-      mdl_ticket->downgrade_exclusive_lock();
+      mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
   }
 
   DBUG_PRINT("exit", ("error: %d", error));
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index ccfe21d1af5..5f800ea8642 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -315,8 +315,11 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
   }
   thd->open_tables= backup_open_tables;
   if (hash_tables->mdl_request.ticket)
+  {
     thd->mdl_context.
-      move_ticket_after_lt_or_ha_sentinel(hash_tables->mdl_request.ticket);
+      move_ticket_after_trans_sentinel(hash_tables->mdl_request.ticket);
+    thd->mdl_context.set_needs_thr_lock_abort(TRUE);
+  }
 
   /* Assert that the above check prevent opening of views and merge tables. */
   DBUG_ASSERT(hash_tables->table->next == NULL);
@@ -376,6 +379,13 @@ bool mysql_ha_close(THD *thd, TABLE_LIST *tables)
     DBUG_RETURN(TRUE);
   }
 
+  /*
+    Mark MDL_context as no longer breaking protocol if we have
+    closed last HANDLER.
+  */
+  if (! thd->handler_tables_hash.records)
+    thd->mdl_context.set_needs_thr_lock_abort(FALSE);
+
   my_ok(thd);
   DBUG_PRINT("exit", ("OK"));
   DBUG_RETURN(FALSE);
@@ -739,6 +749,13 @@ void mysql_ha_rm_tables(THD *thd, TABLE_LIST *tables)
     hash_tables= next;
   }
 
+  /*
+    Mark MDL_context as no longer breaking protocol if we have
+    closed last HANDLER.
+  */
+  if (! thd->handler_tables_hash.records)
+    thd->mdl_context.set_needs_thr_lock_abort(FALSE);
+
   DBUG_VOID_RETURN;
 }
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 397674471c5..0c3ccf0af85 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1972,7 +1972,7 @@ mysql_execute_command(THD *thd)
   */
   if (((sql_command_flags[lex->sql_command] & CF_PROTECT_AGAINST_GRL) != 0) &&
       !thd->locked_tables_mode)
-    if (wait_if_global_read_lock(thd, FALSE, TRUE))
+    if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
       goto error;
 
   switch (lex->sql_command) {
@@ -2045,7 +2045,7 @@ mysql_execute_command(THD *thd)
       break;
 
     if (!thd->locked_tables_mode && lex->protect_against_global_read_lock &&
-        wait_if_global_read_lock(thd, 0, 1))
+        thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
       break;
 
     res= execute_sqlcom_select(thd, all_tables);
@@ -2397,8 +2397,7 @@ case SQLCOM_PREPARE:
         goto end_with_restore_list;
       }
 
-      if (!(res= open_and_lock_tables_derived(thd, lex->query_tables, TRUE,
-                                              MYSQL_OPEN_TAKE_UPGRADABLE_MDL)))
+      if (!(res= open_and_lock_tables_derived(thd, lex->query_tables, TRUE, 0)))
       {
         /*
           Is table which we are changing used somewhere in other parts
@@ -2548,7 +2547,7 @@ end_with_restore_list:
     client thread has locked tables
   */
   if (thd->locked_tables_mode ||
-      thd->active_transaction() || thd->global_read_lock)
+      thd->active_transaction() || thd->global_read_lock.is_acquired())
   {
     my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
@@ -3078,7 +3077,7 @@ end_with_restore_list:
                  ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
       goto error;
     }
-    if (wait_if_global_read_lock(thd, 0, 1))
+    if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
       goto error;
     res= mysql_truncate(thd, first_table, 0);
     break;
@@ -3299,8 +3298,8 @@ end_with_restore_list:
       thd->mdl_context.release_transactional_locks();
       thd->options&= ~(OPTION_TABLE_LOCK);
     }
-    if (thd->global_read_lock)
-      unlock_global_read_lock(thd);
+    if (thd->global_read_lock.is_acquired())
+      thd->global_read_lock.unlock_global_read_lock(thd);
     my_ok(thd);
     break;
   case SQLCOM_LOCK_TABLES:
@@ -3322,7 +3321,7 @@ end_with_restore_list:
                            FALSE, UINT_MAX, FALSE))
       goto error;
     if (lex->protect_against_global_read_lock &&
-        wait_if_global_read_lock(thd, 0, 1))
+        thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
       goto error;
 
     init_mdl_requests(all_tables);
@@ -4503,13 +4502,13 @@ error:
   res= TRUE;
 
 finish:
-  if (thd->global_read_lock_protection > 0)
+  if (thd->global_read_lock.has_protection())
   {
     /*
       Release the protection against the global read lock and wake
       everyone, who might want to set a global read lock.
     */
-    start_waiting_global_read_lock(thd);
+    thd->global_read_lock.start_waiting_global_read_lock(thd);
   }
 
   if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END))
@@ -5971,7 +5970,9 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
   ptr->next_name_resolution_table= NULL;
   /* Link table in global list (all used tables) */
   lex->add_to_query_tables(ptr);
-  ptr->mdl_request.init(MDL_key::TABLE, ptr->db, ptr->table_name, MDL_SHARED);
+  ptr->mdl_request.init(MDL_key::TABLE, ptr->db, ptr->table_name,
+                        (ptr->lock_type >= TL_WRITE_ALLOW_WRITE) ?
+                        MDL_SHARED_WRITE : MDL_SHARED_READ);
   DBUG_RETURN(ptr);
 }
 
@@ -6207,6 +6208,8 @@ void st_select_lex::set_lock_for_tables(thr_lock_type lock_type)
   {
     tables->lock_type= lock_type;
     tables->updating=  for_update;
+    tables->mdl_request.set_type((lock_type >= TL_WRITE_ALLOW_WRITE) ?
+                                 MDL_SHARED_WRITE : MDL_SHARED_READ);
   }
   DBUG_VOID_RETURN;
 }
@@ -6503,7 +6506,7 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
   DBUG_ASSERT(!thd || thd->locked_tables_mode ||
               !thd->mdl_context.has_locks() ||
               thd->handler_tables_hash.records ||
-              thd->global_read_lock);
+              thd->global_read_lock.is_acquired());
 
   /*
     Note that if REFRESH_READ_LOCK bit is set then REFRESH_TABLES is set too
@@ -6529,16 +6532,16 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
 	UNLOCK TABLES
       */
       tmp_write_to_binlog= 0;
-      if (lock_global_read_lock(thd))
+      if (thd->global_read_lock.lock_global_read_lock(thd))
 	return 1;                               // Killed
       if (close_cached_tables(thd, tables, FALSE, (options & REFRESH_FAST) ?
                               FALSE : TRUE))
           result= 1;
       
-      if (make_global_read_lock_block_commit(thd)) // Killed
+      if (thd->global_read_lock.make_global_read_lock_block_commit(thd)) // Killed
       {
         /* Don't leave things in a half-locked state */
-        unlock_global_read_lock(thd);
+        thd->global_read_lock.unlock_global_read_lock(thd);
         return 1;
       }
     }
@@ -6553,15 +6556,15 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
         if (tables)
         {
           for (TABLE_LIST *t= tables; t; t= t->next_local)
-            if (!find_write_locked_table(thd->open_tables, t->db,
-                                         t->table_name))
+            if (!find_table_for_mdl_upgrade(thd->open_tables, t->db,
+                                            t->table_name, FALSE))
               return 1;
         }
         else
         {
           for (TABLE *tab= thd->open_tables; tab; tab= tab->next)
           {
-            if (tab->reginfo.lock_type < TL_WRITE_ALLOW_WRITE)
+            if (! tab->mdl_ticket->is_upgradable_or_exclusive())
             {
               my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0),
                        tab->s->table_name.str);
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 70f69c75de3..031cb846be4 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1204,7 +1204,8 @@ static bool mysql_test_insert(Prepared_statement *stmt,
     If we would use locks, then we have to ensure we are not using
     TL_WRITE_DELAYED as having two such locks can cause table corruption.
   */
-  if (open_normal_and_derived_tables(thd, table_list, 0))
+  if (open_normal_and_derived_tables(thd, table_list,
+                                     MYSQL_OPEN_FORCE_SHARED_MDL))
     goto error;
 
   if ((values= its++))
@@ -1285,7 +1286,7 @@ static int mysql_test_update(Prepared_statement *stmt,
   DBUG_ENTER("mysql_test_update");
 
   if (update_precheck(thd, table_list) ||
-      open_tables(thd, &table_list, &table_count, 0))
+      open_tables(thd, &table_list, &table_count, MYSQL_OPEN_FORCE_SHARED_MDL))
     goto error;
 
   if (table_list->multitable_view)
@@ -1362,7 +1363,8 @@ static bool mysql_test_delete(Prepared_statement *stmt,
   DBUG_ENTER("mysql_test_delete");
 
   if (delete_precheck(thd, table_list) ||
-      open_normal_and_derived_tables(thd, table_list, 0))
+      open_normal_and_derived_tables(thd, table_list,
+                                     MYSQL_OPEN_FORCE_SHARED_MDL))
     goto error;
 
   if (!table_list->table)
@@ -1420,7 +1422,7 @@ static int mysql_test_select(Prepared_statement *stmt,
     goto error;
   }
 
-  if (open_normal_and_derived_tables(thd, tables, 0))
+  if (open_normal_and_derived_tables(thd, tables, MYSQL_OPEN_FORCE_SHARED_MDL))
     goto error;
 
   thd->used_tables= 0;                        // Updated by setup_fields
@@ -1481,7 +1483,7 @@ static bool mysql_test_do_fields(Prepared_statement *stmt,
                                    UINT_MAX, FALSE))
     DBUG_RETURN(TRUE);
 
-  if (open_normal_and_derived_tables(thd, tables, 0))
+  if (open_normal_and_derived_tables(thd, tables, MYSQL_OPEN_FORCE_SHARED_MDL))
     DBUG_RETURN(TRUE);
   DBUG_RETURN(setup_fields(thd, 0, *values, MARK_COLUMNS_NONE, 0, 0));
 }
@@ -1511,7 +1513,7 @@ static bool mysql_test_set_fields(Prepared_statement *stmt,
 
   if ((tables && check_table_access(thd, SELECT_ACL, tables, FALSE,
                                     UINT_MAX, FALSE)) ||
-      open_normal_and_derived_tables(thd, tables, 0))
+      open_normal_and_derived_tables(thd, tables, MYSQL_OPEN_FORCE_SHARED_MDL))
     goto error;
 
   while ((var= it++))
@@ -1548,7 +1550,7 @@ static bool mysql_test_call_fields(Prepared_statement *stmt,
 
   if ((tables && check_table_access(thd, SELECT_ACL, tables, FALSE,
                                     UINT_MAX, FALSE)) ||
-      open_normal_and_derived_tables(thd, tables, 0))
+      open_normal_and_derived_tables(thd, tables, MYSQL_OPEN_FORCE_SHARED_MDL))
     goto err;
 
   while ((item= it++))
@@ -1631,7 +1633,8 @@ select_like_stmt_test_with_open(Prepared_statement *stmt,
     prepared EXPLAIN yet so derived tables will clean up after
     themself.
   */
-  if (open_normal_and_derived_tables(stmt->thd, tables, 0))
+  if (open_normal_and_derived_tables(stmt->thd, tables,
+                                     MYSQL_OPEN_FORCE_SHARED_MDL))
     DBUG_RETURN(TRUE);
 
   DBUG_RETURN(select_like_stmt_test(stmt, specific_prepare,
@@ -1675,7 +1678,8 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
 
   if (select_lex->item_list.elements)
   {
-    if (open_normal_and_derived_tables(stmt->thd, lex->query_tables, 0))
+    if (open_normal_and_derived_tables(stmt->thd, lex->query_tables,
+                                       MYSQL_OPEN_FORCE_SHARED_MDL))
       DBUG_RETURN(TRUE);
 
     select_lex->context.resolve_in_select_list= TRUE;
@@ -1694,7 +1698,8 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
       we validate metadata of all CREATE TABLE statements,
       which keeps metadata validation code simple.
     */
-    if (open_normal_and_derived_tables(stmt->thd, lex->query_tables, 0))
+    if (open_normal_and_derived_tables(stmt->thd, lex->query_tables,
+                                       MYSQL_OPEN_FORCE_SHARED_MDL))
       DBUG_RETURN(TRUE);
   }
 
@@ -1727,7 +1732,7 @@ static bool mysql_test_create_view(Prepared_statement *stmt)
   if (create_view_precheck(thd, tables, view, lex->create_view_mode))
     goto err;
 
-  if (open_normal_and_derived_tables(thd, tables, 0))
+  if (open_normal_and_derived_tables(thd, tables, MYSQL_OPEN_FORCE_SHARED_MDL))
     goto err;
 
   lex->view_prepare_mode= 1;
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index ca1543b32d8..50cd361b174 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -53,7 +53,7 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
 
   mysql_ha_rm_tables(thd, table_list);
 
-  if (wait_if_global_read_lock(thd,0,1))
+  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
     DBUG_RETURN(1);
 
   if (logger.is_log_table_enabled(QUERY_LOG_GENERAL) ||
@@ -189,7 +189,7 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
   unlock_table_names(thd);
 
 err:
-  start_waiting_global_read_lock(thd);
+  thd->global_read_lock.start_waiting_global_read_lock(thd);
   DBUG_RETURN(error);
 }
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 278e0c1445f..9a4ce112ffe 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3085,12 +3085,10 @@ try_acquire_high_prio_shared_mdl_lock(THD *thd, TABLE_LIST *table,
   table->mdl_request.init(MDL_key::TABLE, table->db, table->table_name,
                           MDL_SHARED_HIGH_PRIO);
   while (!(error=
-           thd->mdl_context.try_acquire_shared_lock(&table->mdl_request)) &&
+           thd->mdl_context.try_acquire_lock(&table->mdl_request)) &&
          !table->mdl_request.ticket && !can_deadlock)
   {
-    MDL_request_list mdl_requests;
-    mdl_requests.push_front(&table->mdl_request);
-    if ((error= thd->mdl_context.wait_for_locks(&mdl_requests)))
+    if ((error= thd->mdl_context.wait_for_lock(&table->mdl_request)))
       break;
   }
   return error;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index e8c2af4c87b..971a1d79363 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1790,7 +1790,8 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
 
   if (!drop_temporary)
   {
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
+    if (!thd->locked_tables_mode &&
+       thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
       DBUG_RETURN(TRUE);
   }
 
@@ -1798,8 +1799,8 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   error= mysql_rm_table_part2(thd, tables, if_exists, drop_temporary, 0, 0);
   thd->pop_internal_handler();
 
-  if (thd->global_read_lock_protection > 0)
-    start_waiting_global_read_lock(thd);
+  if (thd->global_read_lock.has_protection())
+    thd->global_read_lock.start_waiting_global_read_lock(thd);
 
   if (error)
     DBUG_RETURN(TRUE);
@@ -1931,8 +1932,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
             by parser) it is safe to cache pointer to the TABLE instances
             in its elements.
           */
-          table->table= find_write_locked_table(thd->open_tables, table->db,
-                                                table->table_name);
+          table->table= find_table_for_mdl_upgrade(thd->open_tables, table->db,
+                                                   table->table_name, FALSE);
           if (!table->table)
             DBUG_RETURN(1);
           table->mdl_request.ticket= table->table->mdl_ticket;
@@ -2200,7 +2201,7 @@ err:
       Under LOCK TABLES we should release meta-data locks on the tables
       which were dropped. Otherwise we can rely on close_thread_tables()
       doing this. Unfortunately in this case we are likely to get more
-      false positives in try_acquire_exclusive_lock() function. So
+      false positives in try_acquire_lock() function. So
       it makes sense to remove exclusive meta-data locks in all cases.
 
       Leave LOCK TABLES mode if we managed to drop all tables which were
@@ -4142,7 +4143,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
     Open or obtain an exclusive metadata lock on table being created.
   */
   if (open_and_lock_tables_derived(thd, thd->lex->query_tables, FALSE,
-                                   MYSQL_OPEN_TAKE_UPGRADABLE_MDL))
+                                   0))
   {
     result= TRUE;
     goto unlock;
@@ -4353,6 +4354,10 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
 
   if (!(table= table_list->table))
   {
+    char key[MAX_DBKEY_LENGTH];
+    uint key_length;
+    MDL_request mdl_global_request;
+    MDL_request_list mdl_requests;
     /*
       If the table didn't exist, we have a shared metadata lock
       on it that is left from mysql_admin_table()'s attempt to 
@@ -4365,22 +4370,17 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
       Attempt to do full-blown table open in mysql_admin_table() has failed.
       Let us try to open at least a .FRM for this table.
     */
-    char key[MAX_DBKEY_LENGTH];
-    uint key_length;
 
     key_length= create_table_def_key(thd, key, table_list, 0);
     table_list->mdl_request.init(MDL_key::TABLE,
                                  table_list->db, table_list->table_name,
                                  MDL_EXCLUSIVE);
 
-    MDL_request mdl_global_request;
     mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+    mdl_requests.push_front(&table_list->mdl_request);
+    mdl_requests.push_front(&mdl_global_request);
 
-    if (thd->mdl_context.acquire_global_intention_exclusive_lock(
-               &mdl_global_request))
-      DBUG_RETURN(0);
-
-    if (thd->mdl_context.acquire_exclusive_lock(&table_list->mdl_request))
+    if (thd->mdl_context.acquire_locks(&mdl_requests))
       DBUG_RETURN(0);
     has_mdl_lock= TRUE;
 
@@ -5274,8 +5274,9 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
     non-temporary table.
   */
   DBUG_ASSERT((create_info->options & HA_LEX_CREATE_TMP_TABLE) ||
-              thd->mdl_context.is_exclusive_lock_owner(MDL_key::TABLE, table->db,
-                                                       table->table_name));
+              thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
+                                             table->table_name,
+                                             MDL_EXCLUSIVE));
 
   /*
     We have to write the query before we unlock the tables.
@@ -6458,7 +6459,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       DBUG_RETURN(TRUE);
     }
 
-    if (wait_if_global_read_lock(thd,0,1))
+    if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
       DBUG_RETURN(TRUE);
     if (lock_table_names(thd, table_list))
     {
@@ -6484,7 +6485,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     unlock_table_names(thd);
 
 view_err:
-    start_waiting_global_read_lock(thd);
+    thd->global_read_lock.start_waiting_global_read_lock(thd);
     DBUG_RETURN(error);
   }
 
@@ -6568,10 +6569,11 @@ view_err:
           Global intention exclusive lock must have been already acquired when
           table to be altered was open, so there is no need to do it here.
         */
-        DBUG_ASSERT(thd->
-                    mdl_context.is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
+        DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::GLOBAL,
+                                                   "", "",
+                                                   MDL_INTENTION_EXCLUSIVE));
 
-        if (thd->mdl_context.try_acquire_exclusive_lock(&target_mdl_request))
+        if (thd->mdl_context.try_acquire_lock(&target_mdl_request))
           DBUG_RETURN(TRUE);
         if (target_mdl_request.ticket == NULL)
         {
@@ -6767,9 +6769,6 @@ view_err:
         Under LOCK TABLES we should adjust meta-data locks before finishing
         statement. Otherwise we can rely on close_thread_tables() releasing
         them.
-
-        TODO: Investigate what should be done with upgraded table-level
-        lock here...
       */
       if (new_name != table_name || new_db != db)
       {
@@ -6777,7 +6776,7 @@ view_err:
         thd->mdl_context.release_all_locks_for_name(mdl_ticket);
       }
       else
-        mdl_ticket->downgrade_exclusive_lock();
+        mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
     }
     DBUG_RETURN(error);
   }
@@ -7462,7 +7461,7 @@ view_err:
       thd->mdl_context.release_all_locks_for_name(mdl_ticket);
     }
     else
-      mdl_ticket->downgrade_exclusive_lock();
+      mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
   }
 
 end_temporary:
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 8719938d85d..12180d9447c 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -385,7 +385,8 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
     LOCK_open is not enough because global read lock is held without holding
     LOCK_open).
   */
-  if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
+  if (!thd->locked_tables_mode &&
+      thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
     DBUG_RETURN(TRUE);
 
   if (!create)
@@ -453,8 +454,10 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   if (thd->locked_tables_mode)
   {
     /* Under LOCK TABLES we must only accept write locked tables. */
-    if (!(tables->table= find_write_locked_table(thd->open_tables, tables->db,
-                                                 tables->table_name)))
+    if (!(tables->table= find_table_for_mdl_upgrade(thd->open_tables,
+                                                    tables->db,
+                                                    tables->table_name,
+                                                    FALSE)))
       goto end;
   }
   else
@@ -517,10 +520,10 @@ end:
     TABLE instance created by open_n_lock_single_table() and metadata lock.
   */
   if (thd->locked_tables_mode && tables && lock_upgrade_done)
-    mdl_ticket->downgrade_exclusive_lock();
+    mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
 
-  if (thd->global_read_lock_protection > 0)
-    start_waiting_global_read_lock(thd);
+  if (thd->global_read_lock.has_protection())
+    thd->global_read_lock.start_waiting_global_read_lock(thd);
 
   if (!result)
     my_ok(thd);
@@ -1886,7 +1889,8 @@ bool Table_triggers_list::change_table_name(THD *thd, const char *db,
     In the future, only an exclusive metadata lock will be enough.
   */
 #ifndef DBUG_OFF
-  if (thd->mdl_context.is_exclusive_lock_owner(MDL_key::TABLE, db, old_table))
+  if (thd->mdl_context.is_lock_owner(MDL_key::TABLE, db, old_table,
+                                     MDL_EXCLUSIVE))
     safe_mutex_assert_owner(&LOCK_open);
 #endif
 
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 51f80c589b2..c9b6a926649 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -614,7 +614,7 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
 #endif
 
 
-  if (wait_if_global_read_lock(thd, 0, 0))
+  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
   {
     res= TRUE;
     goto err;
@@ -669,7 +669,7 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
   pthread_mutex_unlock(&LOCK_open);
   if (mode != VIEW_CREATE_NEW)
     query_cache_invalidate3(thd, view, 0);
-  start_waiting_global_read_lock(thd);
+  thd->global_read_lock.start_waiting_global_read_lock(thd);
   if (res)
     goto err;
 
diff --git a/sql/table.cc b/sql/table.cc
index 74add641544..2bd7a8cc7ec 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -4583,11 +4583,12 @@ void TABLE_LIST::reinit_before_use(THD *thd)
 
   mdl_request.ticket= NULL;
   /*
-    Not strictly necessary, but we manipulate with the type in open_table(),
-    so it's "safe" to reset the lock request type to the parser default, to
-    restore things back to first-execution state.
+    Since we manipulate with the metadata lock type in open_table(),
+    we need to reset it to the parser default, to restore things back
+    to first-execution state.
   */
-  mdl_request.set_type(MDL_SHARED);
+  mdl_request.set_type((lock_type >= TL_WRITE_ALLOW_WRITE) ?
+                       MDL_SHARED_WRITE : MDL_SHARED_READ);
 }
 
 /*
@@ -4821,7 +4822,8 @@ void init_mdl_requests(TABLE_LIST *table_list)
   for ( ; table_list ; table_list= table_list->next_global)
     table_list->mdl_request.init(MDL_key::TABLE,
                                  table_list->db, table_list->table_name,
-                                 MDL_SHARED);
+                                 table_list->lock_type >= TL_WRITE_ALLOW_WRITE ?
+                                 MDL_SHARED_WRITE : MDL_SHARED_READ);
 }
 
 
diff --git a/sql/table.h b/sql/table.h
index fb9c335d877..6f75e693021 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1103,7 +1103,9 @@ struct TABLE_LIST
     table_name_length= table_name_length_arg;
     alias= (char*) alias_arg;
     lock_type= lock_type_arg;
-    mdl_request.init(MDL_key::TABLE, db, table_name, MDL_SHARED);
+    mdl_request.init(MDL_key::TABLE, db, table_name,
+                     (lock_type >= TL_WRITE_ALLOW_WRITE) ?
+                     MDL_SHARED_WRITE : MDL_SHARED_READ);
   }
 
   /*
diff --git a/sql/transaction.cc b/sql/transaction.cc
index 1ca455028f0..7b34826d154 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -586,7 +586,7 @@ bool trans_xa_commit(THD *thd)
   }
   else if (xa_state == XA_PREPARED && thd->lex->xa_opt == XA_NONE)
   {
-    if (wait_if_global_read_lock(thd, 0, 0))
+    if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, FALSE))
     {
       ha_rollback_trans(thd, TRUE);
       my_error(ER_XAER_RMERR, MYF(0));
@@ -596,7 +596,7 @@ bool trans_xa_commit(THD *thd)
       res= test(ha_commit_one_phase(thd, 1));
       if (res)
         my_error(ER_XAER_RMERR, MYF(0));
-      start_waiting_global_read_lock(thd);
+      thd->global_read_lock.start_waiting_global_read_lock(thd);
     }
   }
   else