Backport of:

------------------------------------------------------------
revno: 2617.23.18
committer: Davi Arnaut <Davi.Arnaut@Sun.COM>
branch nick: 4284-6.0
timestamp: Mon 2009-03-02 18:18:26 -0300
message:
Bug#989: If DROP TABLE while there's an active transaction, wrong binlog order
WL#4284: Transactional DDL locking

This is a prerequisite patch:

These changes are intended to split lock requests from granted
locks and to allow the memory and lifetime of granted locks to
be managed within the MDL subsystem. Furthermore, tickets can
now be shared and therefore are used to satisfy multiple lock
requests, but only shared locks can be recursive.

The problem is that the MDL subsystem morphs lock requests into
granted locks locks but does not manage the memory and lifetime
of lock requests, and hence, does not manage the memory of
granted locks either. This can be problematic because it puts the
burden of tracking references on the users of the subsystem and
it can't be easily done in transactional contexts where the locks
have to be kept around for the duration of a transaction.

Another issue is that recursive locks (when the context trying to
acquire a lock already holds a lock on the same object) requires
that each time the lock is granted, a unique lock request/granted
lock structure structure must be kept around until the lock is
released. This can lead to memory leaks in transactional contexts
as locks taken during the transaction should only be released at
the end of the transaction. This also leads to unnecessary wake
ups (broadcasts) in the MDL subsystem if the context still holds
a equivalent of the lock being released.

These issues are exacerbated due to the fact that WL#4284 low-level
design says that the implementation should "2) Store metadata locks
in transaction memory root, rather than statement memory root" but
this is not possible because a memory root, as implemented in mysys,
requires all objects allocated from it to be freed all at once.

This patch combines review input and significant code contributions
from Konstantin Osipov (kostja) and Dmitri Lenev (dlenev).


mysql-test/r/mdl_sync.result:
  Add test case result.
mysql-test/t/mdl_sync.test:
  Add test case for shared lock upgrade case.
sql/event_db_repository.cc:
  Rename mdl_alloc_lock to mdl_request_alloc.
sql/ha_ndbcluster_binlog.cc:
  Use new function names to initialize MDL lock requests.
sql/lock.cc:
  Rename MDL functions.
sql/log_event.cc:
  The MDL request now holds the table and database name data (MDL_KEY).
sql/mdl.cc:
  Move the MDL key to the MDL_LOCK structure in order to make the
  object suitable for allocation from a fixed-size allocator. This
  allows the simplification of the lists in the MDL_LOCK object,
  which now are just two, one for granted tickets and other for
  waiting (upgraders) tickets.
  
  Recursive requests for a shared lock on the same object can now
  be granted using the same lock ticket. This schema is only used
  for shared locks because that the only case that matters. This
  is used to avoid waste of resources in case a context (connection)
  already holds a shared lock on a object.
sql/mdl.h:
  Introduce a metadata lock object key which is used  to uniquely
  identify lock objects.
  
  Separate the structure used to represent pending lock requests
  from the structure used to represent granted metadata locks.
  
  Rename functions used to manipulate locks requests in order to
  have a more consistent function naming schema.
sql/sp_head.cc:
  Rename mdl_alloc_lock to mdl_request_alloc.
sql/sql_acl.cc:
  Rename alloc_mdl_locks to alloc_mdl_requests.
sql/sql_base.cc:
  Various changes to accommodate that lock requests are separated
  from lock tickets (granted locks).
sql/sql_class.h:
  Last acquired lock before the savepoint was set.
sql/sql_delete.cc:
  Various changes to accommodate that lock requests are separated
  from lock tickets (granted locks).
sql/sql_handler.cc:
  Various changes to accommodate that lock requests are separated
  from lock tickets (granted locks).
sql/sql_insert.cc:
  Rename alloc_mdl_locks to alloc_mdl_requests.
sql/sql_parse.cc:
  Rename alloc_mdl_locks to alloc_mdl_requests.
sql/sql_plist.h:
  Typedef for iterator type.
sql/sql_plugin.cc:
  Rename alloc_mdl_locks to alloc_mdl_requests.
sql/sql_servers.cc:
  Rename alloc_mdl_locks to alloc_mdl_requests.
sql/sql_show.cc:
  Various changes to accommodate that lock requests are separated
  from lock tickets (granted locks).
sql/sql_table.cc:
  Various changes to accommodate that lock requests are separated
  from lock tickets (granted locks).
sql/sql_trigger.cc:
  Save reference to the lock ticket so it can be downgraded later.
sql/sql_udf.cc:
  Rename alloc_mdl_locks to alloc_mdl_requests.
sql/table.cc:
  Rename mdl_alloc_lock to mdl_request_alloc.
sql/table.h:
  Separate MDL lock requests from lock tickets (granted locks).
storage/myisammrg/ha_myisammrg.cc:
  Rename alloc_mdl_locks to alloc_mdl_requests.

1 files changed, 551 insertions, 475 deletions

diff --git a/sql/mdl.cc b/sql/mdl.cc
index 78a85f49952..18f0debdbb5 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -16,6 +16,7 @@
 
 
 #include "mdl.h"
+#include "debug_sync.h"
 #include <hash.h>
 #include <mysqld_error.h>
 
@@ -29,39 +30,43 @@
 
 struct MDL_LOCK
 {
-  I_P_List<MDL_LOCK_DATA, MDL_LOCK_DATA_lock> active_shared;
+  typedef I_P_List<MDL_LOCK_TICKET,
+                   I_P_List_adapter<MDL_LOCK_TICKET,
+                                    &MDL_LOCK_TICKET::next_in_lock,
+                                    &MDL_LOCK_TICKET::prev_in_lock> >
+          Ticket_list;
+
+  typedef Ticket_list::Iterator Ticket_iterator;
+
+  /** The type of lock (shared or exclusive). */
+  enum
+  {
+    SHARED,
+    EXCLUSIVE,
+  } type;
+  /** The key of the object (data) being protected. */
+  MDL_KEY key;
+  /** List of granted tickets for this lock. */
+  Ticket_list granted;
   /*
     There can be several upgraders and active exclusive
     belonging to the same context.
   */
-  I_P_List<MDL_LOCK_DATA, MDL_LOCK_DATA_lock> active_shared_waiting_upgrade;
-  I_P_List<MDL_LOCK_DATA, MDL_LOCK_DATA_lock> active_exclusive;
-  I_P_List<MDL_LOCK_DATA, MDL_LOCK_DATA_lock> waiting_exclusive;
-  /**
-     Number of MDL_LOCK_DATA objects associated with this MDL_LOCK instance
-     and therefore present in one of above lists. Note that this number
-     doesn't account for pending requests for shared lock since we don't
-     associate them with MDL_LOCK and don't keep them in any list.
-  */
-  uint   lock_data_count;
+  Ticket_list waiting;
   void   *cached_object;
   mdl_cached_object_release_hook cached_object_release_hook;
 
-  MDL_LOCK() : cached_object(0), cached_object_release_hook(0) {}
-
-  MDL_LOCK_DATA *get_key_owner()
+  MDL_LOCK(const MDL_KEY *mdl_key)
+  : type(SHARED), cached_object(0), cached_object_release_hook(0)
   {
-     return !active_shared.is_empty() ?
-            active_shared.head() :
-            (!active_shared_waiting_upgrade.is_empty() ?
-             active_shared_waiting_upgrade.head() :
-             (!active_exclusive.is_empty() ?
-              active_exclusive.head() : waiting_exclusive.head()));
+    key.mdl_key_init(mdl_key);
+    granted.empty();
+    waiting.empty();
   }
 
-  bool has_one_lock_data()
+  bool is_empty()
   {
-    return (lock_data_count == 1);
+    return (granted.is_empty() && waiting.is_empty());
   }
 };
 
@@ -86,12 +91,13 @@ struct MDL_GLOBAL_LOCK
 } global_lock;
 
 
-extern "C" uchar *mdl_locks_key(const uchar *record, size_t *length,
-                                my_bool not_used __attribute__((unused)))
+extern "C" uchar *
+mdl_locks_key(const uchar *record, size_t *length,
+              my_bool not_used __attribute__((unused)))
 {
   MDL_LOCK *entry=(MDL_LOCK*) record;
-  *length= entry->get_key_owner()->key_length;
-  return (uchar*) entry->get_key_owner()->key;
+  *length= entry->key.length();
+  return (uchar*) entry->key.ptr();
 }
 
 
@@ -148,7 +154,8 @@ void mdl_destroy()
 
 void mdl_context_init(MDL_CONTEXT *context, THD *thd)
 {
-  context->locks.empty();
+  context->requests.empty();
+  context->tickets.empty();
   context->thd= thd;
   context->has_global_shared_lock= FALSE;
 }
@@ -168,7 +175,8 @@ void mdl_context_init(MDL_CONTEXT *context, THD *thd)
 
 void mdl_context_destroy(MDL_CONTEXT *context)
 {
-  DBUG_ASSERT(context->locks.is_empty());
+  DBUG_ASSERT(context->requests.is_empty());
+  DBUG_ASSERT(context->tickets.is_empty());
   DBUG_ASSERT(!context->has_global_shared_lock);
 }
 
@@ -185,8 +193,10 @@ void mdl_context_destroy(MDL_CONTEXT *context)
 
 void mdl_context_backup_and_reset(MDL_CONTEXT *ctx, MDL_CONTEXT *backup)
 {
-  backup->locks.empty();
-  ctx->locks.swap(backup->locks);
+  backup->requests.empty();
+  backup->tickets.empty();
+  ctx->requests.swap(backup->requests);
+  ctx->tickets.swap(backup->tickets);
 }
 
 
@@ -196,8 +206,10 @@ void mdl_context_backup_and_reset(MDL_CONTEXT *ctx, MDL_CONTEXT *backup)
 
 void mdl_context_restore(MDL_CONTEXT *ctx, MDL_CONTEXT *backup)
 {
-  DBUG_ASSERT(ctx->locks.is_empty());
-  ctx->locks.swap(backup->locks);
+  DBUG_ASSERT(ctx->requests.is_empty());
+  DBUG_ASSERT(ctx->tickets.is_empty());
+  ctx->requests.swap(backup->requests);
+  ctx->tickets.swap(backup->tickets);
 }
 
 
@@ -207,20 +219,29 @@ void mdl_context_restore(MDL_CONTEXT *ctx, MDL_CONTEXT *backup)
 
 void mdl_context_merge(MDL_CONTEXT *dst, MDL_CONTEXT *src)
 {
-  MDL_LOCK_DATA *lock_data;
+  MDL_LOCK_TICKET *ticket;
+  MDL_LOCK_REQUEST *lock_req;
 
   DBUG_ASSERT(dst->thd == src->thd);
 
-  if (!src->locks.is_empty())
+  if (!src->requests.is_empty())
+  {
+    MDL_CONTEXT::Request_iterator it(src->requests);
+    while ((lock_req= it++))
+      dst->requests.push_front(lock_req);
+    src->requests.empty();
+  }
+
+  if (!src->tickets.is_empty())
   {
-    I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(src->locks);
-    while ((lock_data= it++))
+    MDL_CONTEXT::Ticket_iterator it(src->tickets);
+    while ((ticket= it++))
     {
-      DBUG_ASSERT(lock_data->ctx);
-      lock_data->ctx= dst;
-      dst->locks.push_front(lock_data);
+      DBUG_ASSERT(ticket->ctx);
+      ticket->ctx= dst;
+      dst->tickets.push_front(ticket);
     }
-    src->locks.empty();
+    src->tickets.empty();
   }
 }
 
@@ -230,53 +251,36 @@ void mdl_context_merge(MDL_CONTEXT *dst, MDL_CONTEXT *src)
 
    This is to be used for every lock request.
 
-   Note that initialization and allocation are split
-   into two calls. This is to allow flexible memory management
-   of lock requests. Normally a lock request is stored
-   in statement memory (e.g. is a member of struct TABLE_LIST),
-   but we would also like to allow allocation of lock
-   requests in other memory roots, for example in the grant
-   subsystem, to lock privilege tables.
-
-   The MDL subsystem does not own or manage memory of lock
-   requests. Instead it assumes that the life time of every lock
-   request encloses calls to mdl_acquire_shared_lock() and
-   mdl_release_locks().
-
-   @param  lock_data  Pointer to an MDL_LOCK_DATA object to initialize
-   @param  key_buff   Pointer to the buffer for key for the lock request
-                      (should be at least 4+ strlen(db) + 1 + strlen(name)
-                      + 1 bytes, or, if the lengths are not known,
-                      MAX_MDLKEY_LENGTH)
+   Note that initialization and allocation are split into two
+   calls. This is to allow flexible memory management of lock
+   requests. Normally a lock request is stored in statement memory
+   (e.g. is a member of struct TABLE_LIST), but we would also like
+   to allow allocation of lock requests in other memory roots,
+   for example in the grant subsystem, to lock privilege tables.
+
+   The MDL subsystem does not own or manage memory of lock requests.
+   Instead it assumes that the life time of every lock request (including
+   encompassed members db/name) encloses calls to mdl_request_add()
+   and mdl_request_remove() or mdl_request_remove_all().
+
+   @param  lock_req   Pointer to an MDL_LOCK_REQUEST object to initialize
    @param  type       Id of type of object to be locked
    @param  db         Name of database to which the object belongs
    @param  name       Name of of the object
 
-   Stores the database name, object name and the type in the key
-   buffer. Initializes mdl_el to point to the key.
-   We can't simply initialize MDL_LOCK_DATA with type, db and name
-   by-pointer because of the underlying HASH implementation
-   requires the key to be a contiguous buffer.
-
    The initialized lock request will have MDL_SHARED type.
 
    Suggested lock types: TABLE - 0 PROCEDURE - 1 FUNCTION - 2
-   Note that tables and views have the same lock type, since
+   Note that tables and views must have the same lock type, since
    they share the same name space in the SQL standard.
 */
 
-void mdl_init_lock(MDL_LOCK_DATA *lock_data, char *key, int type,
-                   const char *db, const char *name)
+void mdl_request_init(MDL_LOCK_REQUEST *lock_req, unsigned char type,
+                      const char *db, const char *name)
 {
-  int4store(key, type);
-  lock_data->key_length= (uint) (strmov(strmov(key+4, db)+1, name)-key)+1;
-  lock_data->key= key;
-  lock_data->type= MDL_SHARED;
-  lock_data->state= MDL_INITIALIZED;
-#ifndef DBUG_OFF
-  lock_data->ctx= 0;
-  lock_data->lock= 0;
-#endif
+  lock_req->key.mdl_key_init(type, db, name);
+  lock_req->type= MDL_SHARED;
+  lock_req->ticket= NULL;
 }
 
 
@@ -298,19 +302,19 @@ void mdl_init_lock(MDL_LOCK_DATA *lock_data, char *key, int type,
    @retval non-0  Pointer to an object representing a lock request
 */
 
-MDL_LOCK_DATA *mdl_alloc_lock(int type, const char *db, const char *name,
-                              MEM_ROOT *root)
+MDL_LOCK_REQUEST *
+mdl_request_alloc(unsigned char type, const char *db,
+                  const char *name, MEM_ROOT *root)
 {
-  MDL_LOCK_DATA *lock_data;
-  char *key;
+  MDL_LOCK_REQUEST *lock_req;
 
-  if (!multi_alloc_root(root, &lock_data, sizeof(MDL_LOCK_DATA), &key,
-                        MAX_MDLKEY_LENGTH, NULL))
+  if (!(lock_req= (MDL_LOCK_REQUEST*) alloc_root(root,
+                                                 sizeof(MDL_LOCK_REQUEST))))
     return NULL;
 
-  mdl_init_lock(lock_data, key, type, db, name);
+  mdl_request_init(lock_req, type, db, name);
 
-  return lock_data;
+  return lock_req;
 }
 
 
@@ -318,121 +322,122 @@ MDL_LOCK_DATA *mdl_alloc_lock(int type, const char *db, const char *name,
    Add a lock request to the list of lock requests of the context.
 
    The procedure to acquire metadata locks is:
-     - allocate and initialize lock requests (mdl_alloc_lock())
-     - associate them with a context (mdl_add_lock())
-     - call mdl_acquire_shared_lock()/mdl_release_lock() (maybe repeatedly).
+     - allocate and initialize lock requests (mdl_request_alloc())
+     - associate them with a context (mdl_request_add())
+     - call mdl_acquire_shared_lock()/mdl_ticket_release() (maybe repeatedly).
 
    Associates a lock request with the given context.
 
    @param  context    The MDL context to associate the lock with.
                       There should be no more than one context per
                       connection, to avoid deadlocks.
-   @param  lock_data  The lock request to be added.
+   @param  lock_req   The lock request to be added.
 */
 
-void mdl_add_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
+void mdl_request_add(MDL_CONTEXT *context, MDL_LOCK_REQUEST *lock_req)
 {
-  DBUG_ENTER("mdl_add_lock");
-  DBUG_ASSERT(lock_data->state == MDL_INITIALIZED);
-  DBUG_ASSERT(!lock_data->ctx);
-  lock_data->ctx= context;
-  context->locks.push_front(lock_data);
+  DBUG_ENTER("mdl_request_add");
+  DBUG_ASSERT(lock_req->ticket == NULL);
+  context->requests.push_front(lock_req);
   DBUG_VOID_RETURN;
 }
 
 
 /**
-   Remove a lock request from the list of lock requests of the context.
+   Remove a lock request from the list of lock requests.
 
    Disassociates a lock request from the given context.
 
    @param  context    The MDL context to remove the lock from.
-   @param  lock_data  The lock request to be removed.
+   @param  lock_req   The lock request to be removed.
 
-   @pre The lock request being removed should correspond to lock which
+   @pre The lock request being removed should correspond to a ticket that
         was released or was not acquired.
 
-   @note Resets lock request for lock released back to its initial state
+   @note Resets lock request back to its initial state
          (i.e. sets type to MDL_SHARED).
 */
 
-void mdl_remove_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
+void mdl_request_remove(MDL_CONTEXT *context, MDL_LOCK_REQUEST *lock_req)
 {
-  DBUG_ENTER("mdl_remove_lock");
-  DBUG_ASSERT(lock_data->state == MDL_INITIALIZED);
-  DBUG_ASSERT(context == lock_data->ctx);
+  DBUG_ENTER("mdl_request_remove");
   /* Reset lock request back to its initial state. */
-  lock_data->type= MDL_SHARED;
-#ifndef DBUG_OFF
-  lock_data->ctx= 0;
-#endif
-  context->locks.remove(lock_data);
+  lock_req->type= MDL_SHARED;
+  lock_req->ticket= NULL;
+  context->requests.remove(lock_req);
   DBUG_VOID_RETURN;
 }
 
 
 /**
-   Clear all lock requests in the context (clear the context).
-
+   Clear all lock requests in the context.
    Disassociates lock requests from the context.
-   All granted locks must be released prior to calling this
-   function.
-
-   In other words, the expected procedure to release locks is:
-     - mdl_release_locks();
-     - mdl_remove_all_locks();
-
-   We could possibly merge mdl_remove_all_locks() and mdl_release_locks(),
-   but this function comes in handy when we need to back off: in that case
-   we release all the locks acquired so-far but do not free them, since
-   we know that the respective lock requests will be used again.
 
    Also resets lock requests back to their initial state (i.e. MDL_SHARED).
 
    @param context Context to be cleared.
 */
 
-void mdl_remove_all_locks(MDL_CONTEXT *context)
+void mdl_request_remove_all(MDL_CONTEXT *context)
 {
-  MDL_LOCK_DATA *lock_data;
-  I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(context->locks);
-  while ((lock_data= it++))
+  MDL_LOCK_REQUEST *lock_req;
+  MDL_CONTEXT::Request_iterator it(context->requests);
+  while ((lock_req= it++))
   {
     /* Reset lock request back to its initial state. */
-    lock_data->type= MDL_SHARED;
-#ifndef DBUG_OFF
-    lock_data->ctx= 0;
-#endif
+    lock_req->type= MDL_SHARED;
+    lock_req->ticket= NULL;
   }
-  context->locks.empty();
+  context->requests.empty();
 }
 
 
 /**
-   Auxiliary functions needed for creation/destruction of MDL_LOCK
-   objects.
+   Auxiliary functions needed for creation/destruction of MDL_LOCK objects.
 
    @todo This naive implementation should be replaced with one that saves
          on memory allocation by reusing released objects.
 */
 
-static MDL_LOCK* get_lock_object(void)
+static MDL_LOCK* alloc_lock_object(const MDL_KEY *mdl_key)
 {
-  return new MDL_LOCK();
+  return new MDL_LOCK(mdl_key);
 }
 
 
-static void release_lock_object(MDL_LOCK *lock)
+static void free_lock_object(MDL_LOCK *lock)
 {
   delete lock;
 }
 
 
 /**
+   Auxiliary functions needed for creation/destruction of MDL_LOCK_TICKET
+   objects.
+
+   @todo This naive implementation should be replaced with one that saves
+         on memory allocation by reusing released objects.
+*/
+
+static MDL_LOCK_TICKET* alloc_ticket_object(MDL_CONTEXT *context)
+{
+  MDL_LOCK_TICKET *ticket= new MDL_LOCK_TICKET;
+  return ticket;
+}
+
+
+static void free_ticket_object(MDL_LOCK_TICKET *ticket)
+{
+  delete ticket;
+}
+
+
+/**
    Helper functions which simplifies writing various checks and asserts.
 */
 
-static bool is_shared(MDL_LOCK_DATA *lock_data)
+template <typename T>
+static inline bool is_shared(T *lock_data)
 {
   return (lock_data->type < MDL_EXCLUSIVE);
 }
@@ -498,7 +503,7 @@ static inline void mdl_exit_cond(MDL_CONTEXT *context,
          request, implying a form of lock on the global metadata, is
          compatible with the current state of the global metadata lock.
 
-   @param lock_data Request for lock on an individual object, implying a
+   @param lock_req  Request for lock on an individual object, implying a
                     certain kind of global metadata lock.
 
    @retval TRUE  - Lock request can be satisfied
@@ -528,9 +533,9 @@ static inline void mdl_exit_cond(MDL_CONTEXT *context,
         type of locks we don't even have any accounting for them.
 */
 
-static bool can_grant_global_lock(MDL_LOCK_DATA *lock_data)
+static bool can_grant_global_lock(enum_mdl_type type, bool is_upgrade)
 {
-  switch (lock_data->type)
+  switch (type)
   {
   case MDL_SHARED:
   case MDL_SHARED_HIGH_PRIO:
@@ -549,7 +554,7 @@ static bool can_grant_global_lock(MDL_LOCK_DATA *lock_data)
       return TRUE;
     break;
   case MDL_EXCLUSIVE:
-    if (lock_data->state == MDL_PENDING_UPGRADE)
+    if (is_upgrade)
     {
       /*
         We are upgrading MDL_SHARED to MDL_EXCLUSIVE.
@@ -586,7 +591,7 @@ static bool can_grant_global_lock(MDL_LOCK_DATA *lock_data)
    Check if request for the lock can be satisfied given current state of lock.
 
    @param  lock       Lock.
-   @param  lock_data  Request for lock.
+   @param  lock_req   Request for lock.
 
    @retval TRUE   Lock request can be satisfied
    @retval FALSE  There is some conflicting lock.
@@ -612,68 +617,96 @@ static bool can_grant_global_lock(MDL_LOCK_DATA *lock_data)
         being upgraded.
 */
 
-static bool can_grant_lock(MDL_LOCK *lock, MDL_LOCK_DATA *lock_data)
+static bool can_grant_lock(MDL_CONTEXT *ctx, MDL_LOCK *lock,
+                           enum_mdl_type type, bool is_upgrade)
 {
-  switch (lock_data->type)
-  {
+  bool can_grant= FALSE;
+
+  switch (type) {
   case MDL_SHARED:
   case MDL_SHARED_UPGRADABLE:
   case MDL_SHARED_HIGH_PRIO:
-    if ((lock->active_exclusive.is_empty() &&
-         (lock_data->type == MDL_SHARED_HIGH_PRIO ||
-          lock->waiting_exclusive.is_empty() &&
-          lock->active_shared_waiting_upgrade.is_empty())) ||
-        (!lock->active_exclusive.is_empty() &&
-         lock->active_exclusive.head()->ctx == lock_data->ctx))
+    if (lock->type == MDL_LOCK::SHARED)
+    {
+      /* Pending exclusive locks have higher priority over shared locks. */
+      if (lock->waiting.is_empty() || type == MDL_SHARED_HIGH_PRIO)
+        can_grant= TRUE;
+    }
+    else if (lock->granted.head()->ctx == 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 ....
       */
-      return TRUE;
+      can_grant= TRUE;
     }
-    else
-      return FALSE;
     break;
   case MDL_EXCLUSIVE:
-    if (lock_data->state == MDL_PENDING_UPGRADE)
+    if (is_upgrade)
     {
       /* We are upgrading MDL_SHARED to MDL_EXCLUSIVE. */
-      MDL_LOCK_DATA *conf_lock_data;
-      I_P_List_iterator<MDL_LOCK_DATA,
-                        MDL_LOCK_DATA_lock> it(lock->active_shared);
+      MDL_LOCK_TICKET *conf_lock_ticket;
+      MDL_LOCK::Ticket_iterator it(lock->granted);
 
       /*
         There should be no active exclusive locks since we own shared lock
         on the object.
       */
-      DBUG_ASSERT(lock->active_exclusive.is_empty() &&
-                  lock->active_shared_waiting_upgrade.head() == lock_data);
+      DBUG_ASSERT(lock->type == MDL_LOCK::SHARED);
 
-      while ((conf_lock_data= it++))
+      while ((conf_lock_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 (conf_lock_data->ctx != lock_data->ctx)
-          return FALSE;
+        if (conf_lock_ticket->ctx != ctx)
+          break;
       }
-      return TRUE;
+      /* Grant lock if there are no conflicting shared locks. */
+      if (conf_lock_ticket == NULL)
+        can_grant= TRUE;
+      break;
     }
-    else
+    else if (lock->type == MDL_LOCK::SHARED)
     {
-      return (lock->active_exclusive.is_empty() &&
-              lock->active_shared_waiting_upgrade.is_empty() &&
-              lock->active_shared.is_empty());
+      can_grant= lock->granted.is_empty();
     }
     break;
   default:
     DBUG_ASSERT(0);
   }
-  return FALSE;
+  return can_grant;
+}
+
+
+/**
+   Check whether the context already holds a compatible lock ticket
+   on a object. Only shared locks can be recursive.
+
+   @param lock_req  Lock request object for lock to be acquired
+
+   @return A pointer to the lock ticket for the object or NULL otherwise.
+*/
+
+static MDL_LOCK_TICKET *
+mdl_context_find_ticket(MDL_CONTEXT *ctx, MDL_LOCK_REQUEST *lock_req)
+{
+  MDL_LOCK_TICKET *ticket;
+  MDL_CONTEXT::Ticket_iterator it(ctx->tickets);
+
+  DBUG_ASSERT(is_shared(lock_req));
+
+  while ((ticket= it++))
+  {
+    if (lock_req->type == ticket->type &&
+        lock_req->key.is_equal(&ticket->lock->key))
+      break;
+  }
+
+  return ticket;
 }
 
 
@@ -691,7 +724,7 @@ static bool can_grant_lock(MDL_LOCK *lock, MDL_LOCK_DATA *lock_data)
    This function must be called after the lock is added to a context.
 
    @param context    [in]  Context containing request for lock
-   @param lock_data  [in]  Lock request object for lock to be acquired
+   @param lock_req   [in]  Lock request object for lock to be acquired
    @param retry      [out] Indicates that conflicting lock exists and another
                            attempt should be made after releasing all current
                            locks and waiting for conflicting lock go away
@@ -702,81 +735,111 @@ static bool can_grant_lock(MDL_LOCK *lock, MDL_LOCK_DATA *lock_data)
                     In the latter case "retry" parameter is set to TRUE.
 */
 
-bool mdl_acquire_shared_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data,
+bool mdl_acquire_shared_lock(MDL_CONTEXT *context, MDL_LOCK_REQUEST *lock_req,
                              bool *retry)
 {
   MDL_LOCK *lock;
+  MDL_KEY *key= &lock_req->key;
+  MDL_LOCK_TICKET *ticket;
   *retry= FALSE;
 
-  DBUG_ASSERT(is_shared(lock_data) && lock_data->state == MDL_INITIALIZED);
-
-  DBUG_ASSERT(lock_data->ctx == context);
+  DBUG_ASSERT(is_shared(lock_req) && lock_req->ticket == NULL);
 
   safe_mutex_assert_not_owner(&LOCK_open);
 
   if (context->has_global_shared_lock &&
-      lock_data->type == MDL_SHARED_UPGRADABLE)
+      lock_req->type == MDL_SHARED_UPGRADABLE)
   {
     my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
     return TRUE;
   }
 
+  /*
+    Check whether the context already holds a shared lock on the object,
+    and if so, grant the request.
+  */
+  if ((ticket= mdl_context_find_ticket(context, lock_req)))
+  {
+    DBUG_ASSERT(ticket->state == MDL_ACQUIRED);
+    lock_req->ticket= ticket;
+    return FALSE;
+  }
+
   pthread_mutex_lock(&LOCK_mdl);
 
-  if (!can_grant_global_lock(lock_data))
+  if (!can_grant_global_lock(lock_req->type, FALSE))
   {
     pthread_mutex_unlock(&LOCK_mdl);
     *retry= TRUE;
     return TRUE;
   }
 
-  if (!(lock= (MDL_LOCK *)my_hash_search(&mdl_locks, (uchar*)lock_data->key,
-                                         lock_data->key_length)))
+  if (!(ticket= alloc_ticket_object(context)))
   {
-    if (!(lock= get_lock_object()))
-    {
-      pthread_mutex_unlock(&LOCK_mdl);
-      return TRUE;
-    }
-    /*
-      Before inserting MDL_LOCK object into hash we should add at least one
-      MDL_LOCK_DATA to its lists in order to provide key for this element.
-      Thus we can't merge two branches of the above if-statement.
-    */
-    lock->active_shared.push_front(lock_data);
-    lock->lock_data_count= 1;
-    if (my_hash_insert(&mdl_locks, (uchar*)lock))
+    pthread_mutex_unlock(&LOCK_mdl);
+    return TRUE;
+  }
+
+  if (!(lock= (MDL_LOCK*) my_hash_search(&mdl_locks,
+                                         key->ptr(), key->length())))
+  {
+    /* Default lock type is MDL_LOCK::SHARED */
+    lock= alloc_lock_object(key);
+    if (!lock || my_hash_insert(&mdl_locks, (uchar*)lock))
     {
-      release_lock_object(lock);
+      free_lock_object(lock);
+      free_ticket_object(ticket);
       pthread_mutex_unlock(&LOCK_mdl);
       return TRUE;
     }
-    lock_data->state= MDL_ACQUIRED;
-    lock_data->lock= lock;
-    if (lock_data->type == MDL_SHARED_UPGRADABLE)
+  }
+
+  if (can_grant_lock(context, lock, lock_req->type, FALSE))
+  {
+    lock->granted.push_front(ticket);
+    context->tickets.push_front(ticket);
+    ticket->state= MDL_ACQUIRED;
+    lock_req->ticket= ticket;
+    ticket->lock= lock;
+    ticket->type= lock_req->type;
+    ticket->ctx= context;
+    if (lock_req->type == MDL_SHARED_UPGRADABLE)
       global_lock.active_intention_exclusive++;
   }
   else
   {
-    if (can_grant_lock(lock, lock_data))
-    {
-      lock->active_shared.push_front(lock_data);
-      lock->lock_data_count++;
-      lock_data->state= MDL_ACQUIRED;
-      lock_data->lock= lock;
-      if (lock_data->type == MDL_SHARED_UPGRADABLE)
-        global_lock.active_intention_exclusive++;
-    }
-    else
-      *retry= TRUE;
+    /* We can't get here if we allocated a new lock. */
+    DBUG_ASSERT(! lock->is_empty());
+    *retry= TRUE;
+    free_ticket_object(ticket);
   }
+
   pthread_mutex_unlock(&LOCK_mdl);
 
   return *retry;
 }
 
 
-static void release_lock(MDL_LOCK_DATA *lock_data);
+static void release_ticket(MDL_CONTEXT *context, MDL_LOCK_TICKET *ticket);
+
+
+/**
+   Notify a thread holding a shared metadata lock of a pending exclusive lock.
+
+   @param thd               Current thread context
+   @param conf_lock_ticket  Conflicting metadata lock
+
+   @retval TRUE   A thread was woken up
+   @retval FALSE  Lock is not a shared one or no thread was woken up
+*/
+
+static bool notify_shared_lock(THD *thd, MDL_LOCK_TICKET *conf_lock_ticket)
+{
+  bool woke= FALSE;
+  if (conf_lock_ticket->type != MDL_EXCLUSIVE)
+    woke= mysql_notify_thread_having_shared_lock(thd, conf_lock_ticket->ctx->thd);
+  return woke;
+}
 
 
 /**
@@ -796,12 +859,13 @@ static void release_lock(MDL_LOCK_DATA *lock_data);
 
 bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context)
 {
-  MDL_LOCK_DATA *lock_data;
   MDL_LOCK *lock;
   bool signalled= FALSE;
   const char *old_msg;
-  I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(context->locks);
+  MDL_LOCK_REQUEST *lock_req;
+  MDL_LOCK_TICKET *ticket;
   st_my_thread_var *mysys_var= my_thread_var;
+  MDL_CONTEXT::Request_iterator it(context->requests);
 
   safe_mutex_assert_not_owner(&LOCK_open);
 
@@ -815,46 +879,44 @@ bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context)
 
   old_msg= MDL_ENTER_COND(context, mysys_var);
 
-  while ((lock_data= it++))
+  while ((lock_req= it++))
   {
-    DBUG_ASSERT(lock_data->type == MDL_EXCLUSIVE &&
-                lock_data->state == MDL_INITIALIZED);
-    if (!(lock= (MDL_LOCK *) my_hash_search(&mdl_locks, (uchar*)lock_data->key,
-                                            lock_data->key_length)))
+    MDL_KEY *key= &lock_req->key;
+    DBUG_ASSERT(lock_req->type == MDL_EXCLUSIVE &&
+                lock_req->ticket == NULL);
+
+    /* Early allocation: ticket is used as a shortcut to the lock. */
+    if (!(ticket= alloc_ticket_object(context)))
+      goto err;
+
+    if (!(lock= (MDL_LOCK*) my_hash_search(&mdl_locks,
+                                           key->ptr(), key->length())))
     {
-      if (!(lock= get_lock_object()))
-        goto err;
-      /*
-        Again before inserting MDL_LOCK into hash provide key for
-        it by adding MDL_LOCK_DATA to one of its lists.
-      */
-      lock->waiting_exclusive.push_front(lock_data);
-      lock->lock_data_count= 1;
-      if (my_hash_insert(&mdl_locks, (uchar*)lock))
+      lock= alloc_lock_object(key);
+      if (!lock || my_hash_insert(&mdl_locks, (uchar*)lock))
       {
-        release_lock_object(lock);
+        free_ticket_object(ticket);
+        free_lock_object(lock);
         goto err;
       }
-      lock_data->lock= lock;
-      lock_data->state= MDL_PENDING;
-    }
-    else
-    {
-      lock->waiting_exclusive.push_front(lock_data);
-      lock->lock_data_count++;
-      lock_data->lock= lock;
-      lock_data->state= MDL_PENDING;
     }
+
+    lock_req->ticket= ticket;
+    ticket->state= MDL_PENDING;
+    ticket->ctx= context;
+    ticket->lock= lock;
+    ticket->type= lock_req->type;
+    lock->waiting.push_front(ticket);
   }
 
   while (1)
   {
     it.rewind();
-    while ((lock_data= it++))
+    while ((lock_req= it++))
     {
-      lock= lock_data->lock;
+      lock= lock_req->ticket->lock;
 
-      if (!can_grant_global_lock(lock_data))
+      if (!can_grant_global_lock(lock_req->type, FALSE))
       {
         /*
           There is an active or pending global shared lock so we have
@@ -863,27 +925,25 @@ bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context)
         signalled= TRUE;
         break;
       }
-      else if (!can_grant_lock(lock, lock_data))
+      else if (!can_grant_lock(context, lock, lock_req->type, FALSE))
       {
-        MDL_LOCK_DATA *conf_lock_data;
-        I_P_List_iterator<MDL_LOCK_DATA,
-                          MDL_LOCK_DATA_lock> it(lock->active_shared);
+        MDL_LOCK_TICKET *conf_lock_ticket;
+        MDL_LOCK::Ticket_iterator it(lock->granted);
 
-        signalled= !lock->active_exclusive.is_empty() ||
-                   !lock->active_shared_waiting_upgrade.is_empty();
+        signalled= (lock->type == MDL_LOCK::EXCLUSIVE);
 
-        while ((conf_lock_data= it++))
-        {
-          signalled|=
-            mysql_notify_thread_having_shared_lock(context->thd,
-                                                   conf_lock_data->ctx->thd);
-        }
+        while ((conf_lock_ticket= it++))
+          signalled|= notify_shared_lock(context->thd, conf_lock_ticket);
 
         break;
       }
     }
-    if (!lock_data)
+    if (!lock_req)
       break;
+
+    /* There is a shared or exclusive lock on the object. */
+    DEBUG_SYNC(context->thd, "mdl_acquire_exclusive_locks_wait");
+
     if (signalled)
       pthread_cond_wait(&COND_mdl, &LOCK_mdl);
     else
@@ -902,13 +962,16 @@ bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context)
       goto err;
   }
   it.rewind();
-  while ((lock_data= it++))
+  while ((lock_req= it++))
   {
     global_lock.active_intention_exclusive++;
-    lock= lock_data->lock;
-    lock->waiting_exclusive.remove(lock_data);
-    lock->active_exclusive.push_front(lock_data);
-    lock_data->state= MDL_ACQUIRED;
+    ticket= lock_req->ticket;
+    lock= ticket->lock;
+    lock->type= MDL_LOCK::EXCLUSIVE;
+    lock->waiting.remove(ticket);
+    lock->granted.push_front(ticket);
+    context->tickets.push_front(ticket);
+    ticket->state= MDL_ACQUIRED;
     if (lock->cached_object)
       (*lock->cached_object_release_hook)(lock->cached_object);
     lock->cached_object= NULL;
@@ -923,10 +986,20 @@ err:
     Ignore those lock requests which were not made MDL_PENDING.
   */
   it.rewind();
-  while ((lock_data= it++) && lock_data->state == MDL_PENDING)
+  while ((lock_req= it++) && lock_req->ticket)
   {
-    release_lock(lock_data);
-    lock_data->state= MDL_INITIALIZED;
+    ticket= lock_req->ticket;
+    DBUG_ASSERT(ticket->state == MDL_PENDING);
+    lock= ticket->lock;
+    free_ticket_object(ticket);
+    lock->waiting.remove(ticket);
+    /* Reset lock request back to its initial state. */
+    lock_req->ticket= NULL;
+    if (lock->is_empty())
+    {
+      my_hash_delete(&mdl_locks, (uchar *)lock);
+      free_lock_object(lock);
+    }
   }
   /* May be some pending requests for shared locks can be satisfied now. */
   pthread_cond_broadcast(&COND_mdl);
@@ -942,77 +1015,61 @@ err:
    new definition has been constructed.
 
    @param context   Context to which shared lock belongs
-   @param lock_data Satisfied request for shared lock to be upgraded
+   @param ticket    Ticket for shared lock to be upgraded
 
    @note In case of failure to upgrade lock (e.g. because upgrader
          was killed) leaves lock in its original state (locked in
          shared mode).
 
+   @note There can be only one upgrader for a lock or we will have deadlock.
+         This invariant is ensured by code outside of metadata subsystem usually
+         by obtaining some sort of exclusive table-level lock (e.g. TL_WRITE,
+         TL_WRITE_ALLOW_READ) before performing upgrade of metadata lock.
+
    @retval FALSE  Success
    @retval TRUE   Failure (thread was killed)
 */
 
 bool mdl_upgrade_shared_lock_to_exclusive(MDL_CONTEXT *context,
-                                          MDL_LOCK_DATA *lock_data)
+                                          MDL_LOCK_TICKET *ticket)
 {
-  MDL_LOCK *lock;
+  MDL_LOCK *lock= ticket->lock;
   const char *old_msg;
   st_my_thread_var *mysys_var= my_thread_var;
 
   DBUG_ENTER("mdl_upgrade_shared_lock_to_exclusive");
+  DEBUG_SYNC(context->thd, "mdl_upgrade_shared_lock_to_exclusive");
 
   safe_mutex_assert_not_owner(&LOCK_open);
 
-  DBUG_ASSERT(lock_data->state == MDL_ACQUIRED);
-
   /* Allow this function to be called twice for the same lock request. */
-  if (lock_data->type == MDL_EXCLUSIVE)
+  if (ticket->type == MDL_EXCLUSIVE)
     DBUG_RETURN(FALSE);
 
-  DBUG_ASSERT(lock_data->type == MDL_SHARED_UPGRADABLE);
-
-  lock= lock_data->lock;
-
   pthread_mutex_lock(&LOCK_mdl);
 
   old_msg= MDL_ENTER_COND(context, mysys_var);
 
-  lock_data->state= MDL_PENDING_UPGRADE;
-  /* Set type of lock request to the type at which we are aiming. */
-  lock_data->type= MDL_EXCLUSIVE;
-  lock->active_shared.remove(lock_data);
-  /*
-    There can be only one upgrader for this lock or we will have deadlock.
-    This invariant is ensured by code outside of metadata subsystem usually
-    by obtaining some sort of exclusive table-level lock (e.g. TL_WRITE,
-    TL_WRITE_ALLOW_READ) before performing upgrade of metadata lock.
-  */
-  DBUG_ASSERT(lock->active_shared_waiting_upgrade.is_empty());
-  lock->active_shared_waiting_upgrade.push_front(lock_data);
 
   /*
-    Since we should have been already acquired intention exclusive global lock
-    this call is only enforcing asserts.
+    Since we should have already acquired an intention exclusive
+    global lock this call is only enforcing asserts.
   */
-  DBUG_ASSERT(can_grant_global_lock(lock_data));
+  DBUG_ASSERT(can_grant_global_lock(MDL_EXCLUSIVE, TRUE));
 
   while (1)
   {
-    if (can_grant_lock(lock, lock_data))
+    if (can_grant_lock(context, lock, MDL_EXCLUSIVE, TRUE))
       break;
 
     bool signalled= FALSE;
-    MDL_LOCK_DATA *conf_lock_data;
-    I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_lock> it(lock->active_shared);
+    MDL_LOCK_TICKET *conf_lock_ticket;
+    MDL_LOCK::Ticket_iterator it(lock->granted);
 
-    while ((conf_lock_data= it++))
+    while ((conf_lock_ticket= it++))
     {
-      if (conf_lock_data->ctx != context)
-      {
-        signalled|=
-          mysql_notify_thread_having_shared_lock(context->thd,
-                                                 conf_lock_data->ctx->thd);
-      }
+      if (conf_lock_ticket->ctx != context)
+        signalled|= notify_shared_lock(context->thd, conf_lock_ticket);
     }
 
     if (signalled)
@@ -1032,10 +1089,6 @@ bool mdl_upgrade_shared_lock_to_exclusive(MDL_CONTEXT *context,
     }
     if (mysys_var->abort)
     {
-      lock_data->state= MDL_ACQUIRED;
-      lock_data->type= MDL_SHARED_UPGRADABLE;
-      lock->active_shared_waiting_upgrade.remove(lock_data);
-      lock->active_shared.push_front(lock_data);
       /* Pending requests for shared locks can be satisfied now. */
       pthread_cond_broadcast(&COND_mdl);
       MDL_EXIT_COND(context, mysys_var, old_msg);
@@ -1043,9 +1096,9 @@ bool mdl_upgrade_shared_lock_to_exclusive(MDL_CONTEXT *context,
     }
   }
 
-  lock->active_shared_waiting_upgrade.remove(lock_data);
-  lock->active_exclusive.push_front(lock_data);
-  lock_data->state= MDL_ACQUIRED;
+  lock->type= MDL_LOCK::EXCLUSIVE;
+  /* Set the new type of lock in the ticket. */
+  ticket->type= MDL_EXCLUSIVE;
   if (lock->cached_object)
     (*lock->cached_object_release_hook)(lock->cached_object);
   lock->cached_object= 0;
@@ -1069,7 +1122,7 @@ bool mdl_upgrade_shared_lock_to_exclusive(MDL_CONTEXT *context,
    block and wait for the lock if the table already exists.
 
    @param context  [in]  The context containing the lock request
-   @param lock     [in]  The lock request
+   @param lock_req [in]  The lock request
    @param conflict [out] Indicates that conflicting lock exists
 
    @retval TRUE  Failure either conflicting lock exists or some error
@@ -1081,13 +1134,15 @@ bool mdl_upgrade_shared_lock_to_exclusive(MDL_CONTEXT *context,
 */
 
 bool mdl_try_acquire_exclusive_lock(MDL_CONTEXT *context,
-                                    MDL_LOCK_DATA *lock_data,
+                                    MDL_LOCK_REQUEST *lock_req,
                                     bool *conflict)
 {
   MDL_LOCK *lock;
+  MDL_LOCK_TICKET *ticket;
+  MDL_KEY *key= &lock_req->key;
 
-  DBUG_ASSERT(lock_data->type == MDL_EXCLUSIVE &&
-              lock_data->state == MDL_INITIALIZED);
+  DBUG_ASSERT(lock_req->type == MDL_EXCLUSIVE &&
+              lock_req->ticket == NULL);
 
   safe_mutex_assert_not_owner(&LOCK_open);
 
@@ -1095,20 +1150,25 @@ bool mdl_try_acquire_exclusive_lock(MDL_CONTEXT *context,
 
   pthread_mutex_lock(&LOCK_mdl);
 
-  if (!(lock= (MDL_LOCK *)my_hash_search(&mdl_locks, (uchar*)lock_data->key,
-                                         lock_data->key_length)))
+  if (!(lock= (MDL_LOCK*) my_hash_search(&mdl_locks,
+                                         key->ptr(), key->length())))
   {
-    if (!(lock= get_lock_object()))
-      goto err;
-    lock->active_exclusive.push_front(lock_data);
-    lock->lock_data_count= 1;
-    if (my_hash_insert(&mdl_locks, (uchar*)lock))
+    ticket= alloc_ticket_object(context);
+    lock= alloc_lock_object(key);
+    if (!ticket || !lock || my_hash_insert(&mdl_locks, (uchar*)lock))
     {
-      release_lock_object(lock);
+      free_ticket_object(ticket);
+      free_lock_object(lock);
       goto err;
     }
-    lock_data->state= MDL_ACQUIRED;
-    lock_data->lock= lock;
+    lock->type= MDL_LOCK::EXCLUSIVE;
+    lock->granted.push_front(ticket);
+    context->tickets.push_front(ticket);
+    ticket->state= MDL_ACQUIRED;
+    lock_req->ticket= ticket;
+    ticket->ctx= context;
+    ticket->lock= lock;
+    ticket->type= lock_req->type;
     global_lock.active_intention_exclusive++;
     pthread_mutex_unlock(&LOCK_mdl);
     return FALSE;
@@ -1183,9 +1243,9 @@ bool mdl_acquire_global_shared_lock(MDL_CONTEXT *context)
 
 bool mdl_wait_for_locks(MDL_CONTEXT *context)
 {
-  MDL_LOCK_DATA *lock_data;
   MDL_LOCK *lock;
-  I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(context->locks);
+  MDL_LOCK_REQUEST *lock_req;
+  MDL_CONTEXT::Request_iterator it(context->requests);
   const char *old_msg;
   st_my_thread_var *mysys_var= my_thread_var;
 
@@ -1207,22 +1267,23 @@ bool mdl_wait_for_locks(MDL_CONTEXT *context)
     pthread_mutex_lock(&LOCK_mdl);
     old_msg= MDL_ENTER_COND(context, mysys_var);
     it.rewind();
-    while ((lock_data= it++))
+    while ((lock_req= it++))
     {
-      DBUG_ASSERT(lock_data->state == MDL_INITIALIZED);
-      if (!can_grant_global_lock(lock_data))
+      MDL_KEY *key= &lock_req->key;
+      DBUG_ASSERT(lock_req->ticket == NULL);
+      if (!can_grant_global_lock(lock_req->type, FALSE))
         break;
       /*
         To avoid starvation we don't wait if we have a conflict against
         request for MDL_EXCLUSIVE lock.
       */
-      if (is_shared(lock_data) &&
-          (lock= (MDL_LOCK *)my_hash_search(&mdl_locks, (uchar*)lock_data->key,
-                                            lock_data->key_length)) &&
-          !can_grant_lock(lock, lock_data))
+      if (is_shared(lock_req) &&
+          (lock= (MDL_LOCK*) my_hash_search(&mdl_locks, key->ptr(),
+                                            key->length())) &&
+          !can_grant_lock(context, lock, lock_req->type, FALSE))
         break;
     }
-    if (!lock_data)
+    if (!lock_req)
     {
       pthread_mutex_unlock(&LOCK_mdl);
       break;
@@ -1237,58 +1298,48 @@ bool mdl_wait_for_locks(MDL_CONTEXT *context)
 
 /**
    Auxiliary function which allows to release particular lock
-   ownership of which is represented by lock request object.
+   ownership of which is represented by a lock ticket object.
 */
 
-static void release_lock(MDL_LOCK_DATA *lock_data)
+static void release_ticket(MDL_CONTEXT *context, MDL_LOCK_TICKET *ticket)
 {
-  MDL_LOCK *lock;
+  MDL_LOCK *lock= ticket->lock;
+  DBUG_ENTER("release_ticket");
+  DBUG_PRINT("enter", ("db=%s name=%s", lock->key.db_name(),
+                                        lock->key.table_name()));
 
-  DBUG_ENTER("release_lock");
-  DBUG_PRINT("enter", ("db=%s name=%s", lock_data->key + 4,
-                        lock_data->key + 4 + strlen(lock_data->key + 4) + 1));
+  safe_mutex_assert_owner(&LOCK_mdl);
+
+  context->tickets.remove(ticket);
+
+  switch (ticket->type)
+  {
+    case MDL_SHARED_UPGRADABLE:
+      global_lock.active_intention_exclusive--;
+      /* Fallthrough. */
+    case MDL_SHARED:
+    case MDL_SHARED_HIGH_PRIO:
+      lock->granted.remove(ticket);
+      break;
+    case MDL_EXCLUSIVE:
+      lock->type= MDL_LOCK::SHARED;
+      lock->granted.remove(ticket);
+      global_lock.active_intention_exclusive--;
+      break;
+    default:
+      DBUG_ASSERT(0);
+  }
 
-  DBUG_ASSERT(lock_data->state == MDL_PENDING ||
-              lock_data->state == MDL_ACQUIRED);
+  free_ticket_object(ticket);
 
-  lock= lock_data->lock;
-  if (lock->has_one_lock_data())
+  if (lock->is_empty())
   {
     my_hash_delete(&mdl_locks, (uchar *)lock);
     DBUG_PRINT("info", ("releasing cached_object cached_object=%p",
                         lock->cached_object));
     if (lock->cached_object)
       (*lock->cached_object_release_hook)(lock->cached_object);
-    release_lock_object(lock);
-    if (lock_data->state == MDL_ACQUIRED &&
-        (lock_data->type == MDL_EXCLUSIVE ||
-         lock_data->type == MDL_SHARED_UPGRADABLE))
-      global_lock.active_intention_exclusive--;
-  }
-  else
-  {
-    switch (lock_data->type)
-    {
-      case MDL_SHARED_UPGRADABLE:
-        global_lock.active_intention_exclusive--;
-        /* Fallthrough. */
-      case MDL_SHARED:
-      case MDL_SHARED_HIGH_PRIO:
-        lock->active_shared.remove(lock_data);
-        break;
-      case MDL_EXCLUSIVE:
-        if (lock_data->state == MDL_PENDING)
-          lock->waiting_exclusive.remove(lock_data);
-        else
-        {
-          lock->active_exclusive.remove(lock_data);
-          global_lock.active_intention_exclusive--;
-        }
-        break;
-      default:
-        DBUG_ASSERT(0);
-    }
-    lock->lock_data_count--;
+    free_lock_object(lock);
   }
 
   DBUG_VOID_RETURN;
@@ -1307,42 +1358,38 @@ static void release_lock(MDL_LOCK_DATA *lock_data)
                   are associated.
 */
 
-void mdl_release_locks(MDL_CONTEXT *context)
+void mdl_ticket_release_all(MDL_CONTEXT *context)
 {
-  MDL_LOCK_DATA *lock_data;
-  I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(context->locks);
-  DBUG_ENTER("mdl_release_locks");
+  MDL_LOCK_TICKET *ticket;
+  MDL_CONTEXT::Ticket_iterator it(context->tickets);
+  DBUG_ENTER("mdl_ticket_release_all");
 
   safe_mutex_assert_not_owner(&LOCK_open);
 
+  /* Detach lock tickets from the requests for back off. */
+  {
+    MDL_LOCK_REQUEST *lock_req;
+    MDL_CONTEXT::Request_iterator it(context->requests);
+
+    while ((lock_req= it++))
+      lock_req->ticket= NULL;
+  }
+
+  if (context->tickets.is_empty())
+    DBUG_VOID_RETURN;
+
   pthread_mutex_lock(&LOCK_mdl);
-  while ((lock_data= it++))
+  while ((ticket= it++))
   {
-    DBUG_PRINT("info", ("found lock to release lock_data=%p", lock_data));
-    /*
-      Don't call release_lock() for a shared lock if has not been
-      granted. Lock state in this case is MDL_INITIALIZED.
-      We have pending and granted shared locks in the same context
-      when this function is called from the "back-off" path of
-      open_tables().
-    */
-    if (lock_data->state != MDL_INITIALIZED)
-    {
-      release_lock(lock_data);
-      lock_data->state= MDL_INITIALIZED;
-#ifndef DBUG_OFF
-      lock_data->lock= 0;
-#endif
-    }
-    /*
-      We will return lock request to its initial state only in
-      mdl_remove_all_locks() since we need to know type of lock
-      request in mdl_wait_for_locks().
-    */
+    DBUG_PRINT("info", ("found lock to release ticket=%p", ticket));
+    release_ticket(context, ticket);
   }
   /* Inefficient but will do for a while */
   pthread_cond_broadcast(&COND_mdl);
   pthread_mutex_unlock(&LOCK_mdl);
+
+  context->tickets.empty();
+
   DBUG_VOID_RETURN;
 }
 
@@ -1351,20 +1398,17 @@ void mdl_release_locks(MDL_CONTEXT *context)
    Release a lock.
 
    @param context   Context containing lock in question
-   @param lock_data Lock to be released
+   @param ticket    Lock to be released
 
 */
 
-void mdl_release_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
+void mdl_ticket_release(MDL_CONTEXT *context, MDL_LOCK_TICKET *ticket)
 {
+  DBUG_ASSERT(context == ticket->ctx);
   safe_mutex_assert_not_owner(&LOCK_open);
 
   pthread_mutex_lock(&LOCK_mdl);
-  release_lock(lock_data);
-#ifndef DBUG_OFF
-  lock_data->lock= 0;
-#endif
-  lock_data->state= MDL_INITIALIZED;
+  release_ticket(context, ticket);
   pthread_cond_broadcast(&COND_mdl);
   pthread_mutex_unlock(&LOCK_mdl);
 }
@@ -1375,34 +1419,43 @@ void mdl_release_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
    object as this lock request, remove lock requests from the context.
 
    @param context   Context containing locks in question
-   @param lock_data One of the locks for the name/object for which all
+   @param ticket    One of the locks for the name/object for which all
                     locks should be released.
 */
 
-void mdl_release_and_remove_all_locks_for_name(MDL_CONTEXT *context,
-                                               MDL_LOCK_DATA *lock_data)
+void mdl_ticket_release_all_for_name(MDL_CONTEXT *context,
+                                     MDL_LOCK_TICKET *ticket)
 {
   MDL_LOCK *lock;
-  I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(context->locks);
-
-  DBUG_ASSERT(lock_data->state == MDL_ACQUIRED);
 
   /*
-    We can use MDL_LOCK_DATA::lock here to identify other locks for the same
-    object since even altough MDL_LOCK object might be reused for different
+    We can use MDL_LOCK_TICKET::lock here to identify other locks for the same
+    object since even though MDL_LOCK object might be reused for different
     lock after the first lock for this object have been released we can't
     have references to this other MDL_LOCK object in this context.
   */
-  lock= lock_data->lock;
+  lock= ticket->lock;
+
+  /* Remove matching lock requests from the context. */
+  MDL_LOCK_REQUEST *lock_req;
+  MDL_CONTEXT::Request_iterator it_lock_req(context->requests);
 
-  while ((lock_data= it++))
+  while ((lock_req= it_lock_req++))
   {
-    DBUG_ASSERT(lock_data->state == MDL_ACQUIRED);
-    if (lock_data->lock == lock)
-    {
-      mdl_release_lock(context, lock_data);
-      mdl_remove_lock(context, lock_data);
-    }
+    DBUG_ASSERT(lock_req->ticket && lock_req->ticket->state == MDL_ACQUIRED);
+    if (lock_req->ticket->lock == lock)
+      mdl_request_remove(context, lock_req);
+  }
+
+  /* Remove matching lock tickets from the context. */
+  MDL_LOCK_TICKET *lock_tkt;
+  MDL_CONTEXT::Ticket_iterator it_lock_tkt(context->tickets);
+
+  while ((lock_tkt= it_lock_tkt++))
+  {
+    DBUG_ASSERT(lock_tkt->state == MDL_ACQUIRED);
+    if (lock_tkt->lock == lock)
+      mdl_ticket_release(context, lock_tkt);
   }
 }
 
@@ -1411,27 +1464,26 @@ void mdl_release_and_remove_all_locks_for_name(MDL_CONTEXT *context,
    Downgrade an exclusive lock to shared metadata lock.
 
    @param context   A context to which exclusive lock belongs
-   @param lock_data Satisfied request for exclusive lock to be downgraded
+   @param ticket    Ticket for exclusive lock to be downgraded
 */
 
 void mdl_downgrade_exclusive_lock(MDL_CONTEXT *context,
-                                  MDL_LOCK_DATA *lock_data)
+                                  MDL_LOCK_TICKET *ticket)
 {
   MDL_LOCK *lock;
 
-  safe_mutex_assert_not_owner(&LOCK_open);
+  DBUG_ASSERT(context == ticket->ctx);
 
-  DBUG_ASSERT(lock_data->state == MDL_ACQUIRED);
+  safe_mutex_assert_not_owner(&LOCK_open);
 
-  if (is_shared(lock_data))
+  if (is_shared(ticket))
     return;
 
-  lock= lock_data->lock;
+  lock= ticket->lock;
 
   pthread_mutex_lock(&LOCK_mdl);
-  lock->active_exclusive.remove(lock_data);
-  lock_data->type= MDL_SHARED_UPGRADABLE;
-  lock->active_shared.push_front(lock_data);
+  lock->type= MDL_LOCK::SHARED;
+  ticket->type= MDL_SHARED_UPGRADABLE;
   pthread_cond_broadcast(&COND_mdl);
   pthread_mutex_unlock(&LOCK_mdl);
 }
@@ -1469,30 +1521,29 @@ void mdl_release_global_shared_lock(MDL_CONTEXT *context)
            FALSE otherwise.
 */
 
-bool mdl_is_exclusive_lock_owner(MDL_CONTEXT *context, int type,
+bool mdl_is_exclusive_lock_owner(MDL_CONTEXT *context, unsigned char type,
                                  const char *db, const char *name)
 {
-  char key[MAX_MDLKEY_LENGTH];
-  uint key_length;
-  MDL_LOCK_DATA *lock_data;
-  I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(context->locks);
-
-  int4store(key, type);
-  key_length= (uint) (strmov(strmov(key+4, db)+1, name)-key)+1;
-
-  while ((lock_data= it++) &&
-         (lock_data->key_length != key_length ||
-          memcmp(lock_data->key, key, key_length) ||
-          !(lock_data->type == MDL_EXCLUSIVE &&
-            lock_data->state == MDL_ACQUIRED)))
-    continue;
-  return lock_data;
+  MDL_KEY key;
+  MDL_LOCK_TICKET *ticket;
+  MDL_CONTEXT::Ticket_iterator it(context->tickets);
+
+  key.mdl_key_init(type, db, name);
+
+  while ((ticket= it++))
+  {
+    if (ticket->lock->type == MDL_LOCK::EXCLUSIVE &&
+        ticket->lock->key.is_equal(&key))
+      break;
+  }
+
+  return ticket;
 }
 
 
 /**
-   Auxiliary function which allows to check if we some kind of lock on
-   the object.
+   Auxiliary function which allows to check if we have some kind of lock on
+   a object.
 
    @param context Current context
    @param type    Id of object type
@@ -1503,24 +1554,22 @@ bool mdl_is_exclusive_lock_owner(MDL_CONTEXT *context, int type,
            FALSE otherwise.
 */
 
-bool mdl_is_lock_owner(MDL_CONTEXT *context, int type, const char *db,
-                        const char *name)
+bool mdl_is_lock_owner(MDL_CONTEXT *context, unsigned char type,
+                       const char *db, const char *name)
 {
-  char key[MAX_MDLKEY_LENGTH];
-  uint key_length;
-  MDL_LOCK_DATA *lock_data;
-  I_P_List_iterator<MDL_LOCK_DATA, MDL_LOCK_DATA_context> it(context->locks);
+  MDL_KEY key;
+  MDL_LOCK_TICKET *ticket;
+  MDL_CONTEXT::Ticket_iterator it(context->tickets);
 
-  int4store(key, type);
-  key_length= (uint) (strmov(strmov(key+4, db)+1, name)-key)+1;
+  key.mdl_key_init(type, db, name);
 
-  while ((lock_data= it++) &&
-         (lock_data->key_length != key_length ||
-          memcmp(lock_data->key, key, key_length) ||
-          lock_data->state != MDL_ACQUIRED))
-    continue;
+  while ((ticket= it++))
+  {
+    if (ticket->lock->key.is_equal(&key))
+      break;
+  }
 
-  return lock_data;
+  return ticket;
 }
 
 
@@ -1528,21 +1577,20 @@ bool mdl_is_lock_owner(MDL_CONTEXT *context, int type, const char *db,
    Check if we have any pending exclusive locks which conflict with
    existing shared lock.
 
-   @param lock_data Shared lock against which check should be performed.
+   @param ticket Shared lock against which check should be performed.
 
    @return TRUE if there are any conflicting locks, FALSE otherwise.
 */
 
-bool mdl_has_pending_conflicting_lock(MDL_LOCK_DATA *lock_data)
+bool mdl_has_pending_conflicting_lock(MDL_LOCK_TICKET *ticket)
 {
   bool result;
 
-  DBUG_ASSERT(is_shared(lock_data) && lock_data->state == MDL_ACQUIRED);
+  DBUG_ASSERT(is_shared(ticket));
   safe_mutex_assert_not_owner(&LOCK_open);
 
   pthread_mutex_lock(&LOCK_mdl);
-  result= !(lock_data->lock->waiting_exclusive.is_empty() &&
-            lock_data->lock->active_shared_waiting_upgrade.is_empty());
+  result= !ticket->lock->waiting.is_empty();
   pthread_mutex_unlock(&LOCK_mdl);
   return result;
 }
@@ -1551,8 +1599,8 @@ bool mdl_has_pending_conflicting_lock(MDL_LOCK_DATA *lock_data)
 /**
    Associate pointer to an opaque object with a lock.
 
-   @param lock_data     Lock request for the lock with which the
-                        object should be associated.
+   @param ticket        Lock ticket for the lock with which the object
+                        should be associated.
    @param cached_object Pointer to the object
    @param release_hook  Cleanup function to be called when MDL subsystem
                         decides to remove lock or associate another object.
@@ -1576,27 +1624,24 @@ bool mdl_has_pending_conflicting_lock(MDL_LOCK_DATA *lock_data)
   lock on this name is released.
 */
 
-void mdl_set_cached_object(MDL_LOCK_DATA *lock_data, void *cached_object,
+void mdl_set_cached_object(MDL_LOCK_TICKET *ticket, void *cached_object,
                            mdl_cached_object_release_hook release_hook)
 {
+  MDL_LOCK *lock= ticket->lock;
   DBUG_ENTER("mdl_set_cached_object");
-  DBUG_PRINT("enter", ("db=%s name=%s cached_object=%p", lock_data->key + 4,
-                       lock_data->key + 4 + strlen(lock_data->key + 4) + 1,
-                       cached_object));
-
-  DBUG_ASSERT(lock_data->state == MDL_ACQUIRED ||
-              lock_data->state == MDL_PENDING_UPGRADE);
-
+  DBUG_PRINT("enter", ("db=%s name=%s cached_object=%p",
+                        lock->key.db_name(), lock->key.table_name(),
+                        cached_object));
   /*
     TODO: This assumption works now since we do mdl_get_cached_object()
           and mdl_set_cached_object() in the same critical section. Once
           this becomes false we will have to call release_hook here and
           use additional mutex protecting 'cached_object' member.
   */
-  DBUG_ASSERT(!lock_data->lock->cached_object);
+  DBUG_ASSERT(!lock->cached_object);
 
-  lock_data->lock->cached_object= cached_object;
-  lock_data->lock->cached_object_release_hook= release_hook;
+  lock->cached_object= cached_object;
+  lock->cached_object_release_hook= release_hook;
 
   DBUG_VOID_RETURN;
 }
@@ -1605,15 +1650,46 @@ void mdl_set_cached_object(MDL_LOCK_DATA *lock_data, void *cached_object,
 /**
    Get a pointer to an opaque object that associated with the lock.
 
-   @param  lock_data Lock request for the lock with which the object is
-                     associated.
+   @param ticket  Lock ticket for the lock which the object is associated to.
 
    @return Pointer to an opaque object associated with the lock.
 */
 
-void* mdl_get_cached_object(MDL_LOCK_DATA *lock_data)
+void* mdl_get_cached_object(MDL_LOCK_TICKET *ticket)
 {
-  DBUG_ASSERT(lock_data->state == MDL_ACQUIRED ||
-              lock_data->state == MDL_PENDING_UPGRADE);
-  return lock_data->lock->cached_object;
+  return ticket->lock->cached_object;
 }
+
+
+
+/**
+  Releases metadata locks that were acquired after a specific savepoint.
+
+  @note Used to release tickets acquired during a savepoint unit.
+  @note It's safe to iterate and unlock any locks after taken after this
+        savepoint because other statements that take other special locks
+        cause a implicit commit (ie LOCK TABLES).
+
+  @param thd  Current thread
+  @param sv   Savepoint
+*/
+
+void mdl_rollback_to_savepoint(MDL_CONTEXT *ctx,
+                               MDL_LOCK_TICKET *mdl_savepoint)
+{
+  MDL_LOCK_TICKET *mdl_lock_ticket;
+  MDL_CONTEXT::Ticket_iterator it(ctx->tickets);
+  DBUG_ENTER("mdl_rollback_to_savepoint");
+
+  while ((mdl_lock_ticket= it++))
+  {
+    /* Stop when lock was acquired before this savepoint. */
+    if (mdl_lock_ticket == mdl_savepoint)
+      break;
+    mdl_ticket_release(ctx, mdl_lock_ticket);
+  }
+
+  DBUG_VOID_RETURN;
+}
+
+