Backport of:

------------------------------------------------------------
revno: 2630.4.32
committer: Dmitry Lenev <dlenev@mysql.com>
branch nick: mysql-6.0-3726-w2
timestamp: Thu 2008-06-19 16:39:58 +0400
message:
  WL#3726 "DDL locking for all metadata objects".

  After-review fixes in progress.

  Ensure that metadata locking subsystem properly handles
  out-of-memory conditions. Clarified MDL interface by
  separating release of locks and removal of lock requests
  from the context.

sql/lock.cc

@@ -965,7 +965,7 @@ bool lock_table_names(THD *thd, TABLE_LIST *table_list)
     lock_table->mdl_lock_data= mdl_lock_data;
   }
   if (mdl_acquire_exclusive_locks(&thd->mdl_context))
-    return 1;
+    goto end;
   return 0;
 
 end:

sql/mdl.cc

@@ -257,7 +257,7 @@ void mdl_context_merge(MDL_CONTEXT *dst, MDL_CONTEXT *src)
 
    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_el with type, db and name
+   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.
 
@@ -275,7 +275,7 @@ void mdl_init_lock(MDL_LOCK_DATA *lock_data, char *key, int 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_PENDING;
+  lock_data->state= MDL_INITIALIZED;
 #ifndef DBUG_OFF
   lock_data->ctx= 0;
   lock_data->lock= 0;
@@ -336,7 +336,7 @@ MDL_LOCK_DATA *mdl_alloc_lock(int type, const char *db, const char *name,
 void mdl_add_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
 {
   DBUG_ENTER("mdl_add_lock");
-  DBUG_ASSERT(lock_data->state == MDL_PENDING);
+  DBUG_ASSERT(lock_data->state == MDL_INITIALIZED);
   DBUG_ASSERT(!lock_data->ctx);
   lock_data->ctx= context;
   context->locks.push_front(lock_data);
@@ -344,6 +344,36 @@ void mdl_add_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
 }
 
 
+/**
+   Remove a lock request from the list of lock requests of the context.
+
+   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.
+
+   @pre The lock request being removed should correspond to lock which
+        was released or was not acquired.
+
+   @note Resets lock request for lock released back to its initial state
+         (i.e. sets type to MDL_SHARED).
+*/
+
+void mdl_remove_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
+{
+  DBUG_ENTER("mdl_remove_lock");
+  DBUG_ASSERT(lock_data->state == MDL_INITIALIZED);
+  DBUG_ASSERT(context == lock_data->ctx);
+  /* 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);
+  DBUG_VOID_RETURN;
+}
+
+
 /**
    Clear all lock requests in the context (clear the context).
 
@@ -629,7 +659,7 @@ bool mdl_acquire_shared_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data,
   MDL_LOCK *lock;
   *retry= FALSE;
 
-  DBUG_ASSERT(is_shared(lock_data) && lock_data->state == MDL_PENDING);
+  DBUG_ASSERT(is_shared(lock_data) && lock_data->state == MDL_INITIALIZED);
 
   DBUG_ASSERT(lock_data->ctx == context);
 
@@ -654,7 +684,11 @@ bool mdl_acquire_shared_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data,
   if (!(lock= (MDL_LOCK *)my_hash_search(&mdl_locks, (uchar*)lock_data->key,
                                          lock_data->key_length)))
   {
-    lock= get_lock_object();
+    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.
@@ -662,7 +696,12 @@ bool mdl_acquire_shared_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data,
     */
     lock->active_shared.push_front(lock_data);
     lock->lock_data_count= 1;
-    my_hash_insert(&mdl_locks, (uchar*)lock);
+    if (my_hash_insert(&mdl_locks, (uchar*)lock))
+    {
+      release_lock_object(lock);
+      pthread_mutex_unlock(&LOCK_mdl);
+      return TRUE;
+    }
     lock_data->state= MDL_ACQUIRED;
     lock_data->lock= lock;
     if (lock_data->type == MDL_SHARED_UPGRADABLE)
@@ -702,9 +741,6 @@ static void release_lock(MDL_LOCK_DATA *lock_data);
    @param context  A context containing requests for exclusive locks
                    The context may not have other lock requests.
 
-   @note In case of failure (for example, if our thread was killed)
-         resets lock requests back to their initial state (MDL_SHARED)
-
    @retval FALSE  Success
    @retval TRUE   Failure
 */
@@ -735,25 +771,32 @@ bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context)
   while ((lock_data= it++))
   {
     DBUG_ASSERT(lock_data->type == MDL_EXCLUSIVE &&
-                lock_data->state == MDL_PENDING);
+                lock_data->state == MDL_INITIALIZED);
     if (!(lock= (MDL_LOCK *) my_hash_search(&mdl_locks, (uchar*)lock_data->key,
                                             lock_data->key_length)))
     {
-      lock= get_lock_object();
+      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;
-      my_hash_insert(&mdl_locks, (uchar*)lock);
+      if (my_hash_insert(&mdl_locks, (uchar*)lock))
+      {
+        release_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;
     }
   }
 
@@ -806,23 +849,7 @@ bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context)
       pthread_cond_timedwait(&COND_mdl, &LOCK_mdl, &abstime);
     }
     if (thd->killed)
-    {
-      /* Remove our pending lock requests from the locks. */
-      it.rewind();
-      while ((lock_data= it++))
-      {
-        DBUG_ASSERT(lock_data->type == MDL_EXCLUSIVE &&
-                    lock_data->state == MDL_PENDING);
-        release_lock(lock_data);
-        /* Return lock request to its initial state. */
-        lock_data->type= MDL_SHARED;
-        context->locks.remove(lock_data);
-      }
-      /* Pending requests for shared locks can be satisfied now. */
-      pthread_cond_broadcast(&COND_mdl);
-      thd->exit_cond(old_msg);
-      return TRUE;
-    }
+      goto err;
   }
   it.rewind();
   while ((lock_data= it++))
@@ -839,6 +866,22 @@ bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context)
   /* As a side-effect THD::exit_cond() unlocks LOCK_mdl. */
   thd->exit_cond(old_msg);
   return FALSE;
+
+err:
+  /*
+    Remove our pending lock requests from the locks.
+    Ignore those lock requests which were not made MDL_PENDING.
+  */
+  it.rewind();
+  while ((lock_data= it++) && lock_data->state == MDL_PENDING)
+  {
+    release_lock(lock_data);
+    lock_data->state= MDL_INITIALIZED;
+  }
+  /* May be some pending requests for shared locks can be satisfied now. */
+  pthread_cond_broadcast(&COND_mdl);
+  thd->exit_cond(old_msg);
+  return TRUE;
 }
 
 
@@ -976,50 +1019,56 @@ bool mdl_upgrade_shared_lock_to_exclusive(MDL_CONTEXT *context,
 
    @param context  [in]  The context containing the lock request
    @param lock     [in]  The lock request
+   @param conflict [out] Indicates that conflicting lock exists
 
-   @retval FALSE the lock was granted
-   @retval TRUE  there were conflicting locks.
+   @retval TRUE  Failure either conflicting lock exists or some error
+                 occured (probably OOM).
+   @retval FALSE Success, lock was acquired.
 
    FIXME: Compared to lock_table_name_if_not_cached()
           it gives sligthly more false negatives.
 */
 
 bool mdl_try_acquire_exclusive_lock(MDL_CONTEXT *context,
-                                    MDL_LOCK_DATA *lock_data)
+                                    MDL_LOCK_DATA *lock_data,
+                                    bool *conflict)
 {
   MDL_LOCK *lock;
 
   DBUG_ASSERT(lock_data->type == MDL_EXCLUSIVE &&
-              lock_data->state == MDL_PENDING);
+              lock_data->state == MDL_INITIALIZED);
 
   safe_mutex_assert_not_owner(&LOCK_open);
 
+  *conflict= FALSE;
+
   pthread_mutex_lock(&LOCK_mdl);
 
   if (!(lock= (MDL_LOCK *)my_hash_search(&mdl_locks, (uchar*)lock_data->key,
                                          lock_data->key_length)))
   {
-    lock= get_lock_object();
+    if (!(lock= get_lock_object()))
+      goto err;
     lock->active_exclusive.push_front(lock_data);
     lock->lock_data_count= 1;
-    my_hash_insert(&mdl_locks, (uchar*)lock);
+    if (my_hash_insert(&mdl_locks, (uchar*)lock))
+    {
+      release_lock_object(lock);
+      goto err;
+    }
     lock_data->state= MDL_ACQUIRED;
     lock_data->lock= lock;
-    lock= 0;
     global_lock.active_intention_exclusive++;
+    pthread_mutex_unlock(&LOCK_mdl);
+    return FALSE;
   }
-  pthread_mutex_unlock(&LOCK_mdl);
 
-  /*
-    FIXME: We can't leave pending MDL_EXCLUSIVE lock request in the list since
-           for such locks we assume that they have MDL_LOCK_DATA::lock properly set.
-           Long term we should clearly define relation between lock types,
-           presence in the context lists and MDL_LOCK_DATA::lock values.
-  */
-  if (lock)
-    context->locks.remove(lock_data);
+  /* There is some lock for the object. */
+  *conflict= TRUE;
 
-  return lock;
+err:
+  pthread_mutex_unlock(&LOCK_mdl);
+  return TRUE;
 }
 
 
@@ -1111,11 +1160,12 @@ bool mdl_wait_for_locks(MDL_CONTEXT *context)
     it.rewind();
     while ((lock_data= it++))
     {
-      DBUG_ASSERT(lock_data->state == MDL_PENDING);
+      DBUG_ASSERT(lock_data->state == MDL_INITIALIZED);
       if (!can_grant_global_lock(lock_data))
         break;
       /*
-        To avoid starvation we don't wait if we have pending MDL_EXCLUSIVE lock.
+        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,
@@ -1149,6 +1199,9 @@ static void release_lock(MDL_LOCK_DATA *lock_data)
   DBUG_PRINT("enter", ("db=%s name=%s", lock_data->key + 4,
                         lock_data->key + 4 + strlen(lock_data->key + 4) + 1));
 
+  DBUG_ASSERT(lock_data->state == MDL_PENDING ||
+              lock_data->state == MDL_ACQUIRED);
+
   lock= lock_data->lock;
   if (lock->has_one_lock_data())
   {
@@ -1184,7 +1237,6 @@ static void release_lock(MDL_LOCK_DATA *lock_data)
         }
         break;
       default:
-        /* TODO Really? How about problems during lock upgrade ? */
         DBUG_ASSERT(0);
     }
     lock->lock_data_count--;
@@ -1224,10 +1276,10 @@ void mdl_release_locks(MDL_CONTEXT *context)
       lists. Allows us to avoid problems in open_tables() in case of
       back-off
     */
-    if (!(is_shared(lock_data) && lock_data->state == MDL_PENDING))
+    if (lock_data->state != MDL_INITIALIZED)
     {
       release_lock(lock_data);
-      lock_data->state= MDL_PENDING;
+      lock_data->state= MDL_INITIALIZED;
 #ifndef DBUG_OFF
       lock_data->lock= 0;
 #endif
@@ -1247,13 +1299,10 @@ void mdl_release_locks(MDL_CONTEXT *context)
 
 /**
    Release a lock.
-   Removes the lock from the context.
 
    @param context   Context containing lock in question
    @param lock_data Lock to be released
 
-   @note Resets lock request for lock released back to its initial state
-         (i.e. sets type to MDL_SHARED).
 */
 
 void mdl_release_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
@@ -1263,13 +1312,9 @@ void mdl_release_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
   pthread_mutex_lock(&LOCK_mdl);
   release_lock(lock_data);
 #ifndef DBUG_OFF
-  lock_data->ctx= 0;
   lock_data->lock= 0;
 #endif
-  lock_data->state= MDL_PENDING;
-  /* Return lock request to its initial state. */
-  lock_data->type= MDL_SHARED;
-  context->locks.remove(lock_data);
+  lock_data->state= MDL_INITIALIZED;
   pthread_cond_broadcast(&COND_mdl);
   pthread_mutex_unlock(&LOCK_mdl);
 }
@@ -1277,17 +1322,15 @@ void mdl_release_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data)
 
 /**
    Release all locks in the context which correspond to the same name/
-   object as this lock request.
+   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
                     locks should be released.
-
-   @see mdl_release_lock()
 */
 
-void mdl_release_all_locks_for_name(MDL_CONTEXT *context,
-                                    MDL_LOCK_DATA *lock_data)
+void mdl_release_and_remove_all_locks_for_name(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);
@@ -1306,7 +1349,10 @@ void mdl_release_all_locks_for_name(MDL_CONTEXT *context,
   {
     DBUG_ASSERT(lock_data->state == MDL_ACQUIRED);
     if (lock_data->lock == lock)
+    {
       mdl_release_lock(context, lock_data);
+      mdl_remove_lock(context, lock_data);
+    }
   }
 }
 
@@ -1418,9 +1464,10 @@ bool mdl_is_lock_owner(MDL_CONTEXT *context, int type, const char *db,
   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->state == MDL_PENDING))
+  while ((lock_data= it++) &&
+         (lock_data->key_length != key_length ||
+          memcmp(lock_data->key, key, key_length) ||
+          lock_data->state != MDL_ACQUIRED))
     continue;
 
   return lock_data;

sql/mdl.h

@@ -44,7 +44,8 @@ enum enum_mdl_type {MDL_SHARED=0, MDL_SHARED_HIGH_PRIO,
 
 /** States which metadata lock request can have. */
 
-enum enum_mdl_state {MDL_PENDING=0, MDL_ACQUIRED, MDL_PENDING_UPGRADE};
+enum enum_mdl_state {MDL_INITIALIZED=0, MDL_PENDING,
+                     MDL_ACQUIRED, MDL_PENDING_UPGRADE};
 
 
 /**
@@ -152,6 +153,7 @@ void mdl_init_lock(MDL_LOCK_DATA *lock_data, char *key, int type,
 MDL_LOCK_DATA *mdl_alloc_lock(int type, const char *db, const char *name,
                               MEM_ROOT *root);
 void mdl_add_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data);
+void mdl_remove_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data);
 void mdl_remove_all_locks(MDL_CONTEXT *context);
 
 /**
@@ -160,7 +162,7 @@ void mdl_remove_all_locks(MDL_CONTEXT *context);
 
 inline void mdl_set_lock_type(MDL_LOCK_DATA *lock_data, enum_mdl_type lock_type)
 {
-  DBUG_ASSERT(lock_data->state == MDL_PENDING);
+  DBUG_ASSERT(lock_data->state == MDL_INITIALIZED);
   lock_data->type= lock_type;
 }
 
@@ -170,14 +172,15 @@ bool mdl_acquire_exclusive_locks(MDL_CONTEXT *context);
 bool mdl_upgrade_shared_lock_to_exclusive(MDL_CONTEXT *context,
                                           MDL_LOCK_DATA *lock_data);
 bool mdl_try_acquire_exclusive_lock(MDL_CONTEXT *context,
-                                    MDL_LOCK_DATA *lock_data);
+                                    MDL_LOCK_DATA *lock_data,
+                                    bool *conflict);
 bool mdl_acquire_global_shared_lock(MDL_CONTEXT *context);
 
 bool mdl_wait_for_locks(MDL_CONTEXT *context);
 
 void mdl_release_locks(MDL_CONTEXT *context);
-void mdl_release_all_locks_for_name(MDL_CONTEXT *context,
-                                    MDL_LOCK_DATA *lock_data);
+void mdl_release_and_remove_all_locks_for_name(MDL_CONTEXT *context,
+                                               MDL_LOCK_DATA *lock_data);
 void mdl_release_lock(MDL_CONTEXT *context, MDL_LOCK_DATA *lock_data);
 void mdl_downgrade_exclusive_lock(MDL_CONTEXT *context,
                                   MDL_LOCK_DATA *lock_data);

sql/sql_base.cc

@@ -2278,9 +2278,9 @@ void table_share_release_hook(void *share)
 }
 
 
-/*
-  A helper function that acquires an MDL lock for a table
-  being opened.
+/**
+   A helper function that acquires an MDL lock for a table
+   being opened.
 */
 
 static bool
@@ -2304,7 +2304,10 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
     */
     mdl_set_lock_type(mdl_lock_data, MDL_EXCLUSIVE);
     if (mdl_acquire_exclusive_locks(&thd->mdl_context))
+    {
+      mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
       return 1;
+    }
   }
   else
   {
@@ -2327,6 +2330,8 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
     {
       if (retry)
         *action= OT_BACK_OFF_AND_RETRY;
+      else
+        mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
       return 1;
     }
   }
@@ -2833,7 +2838,11 @@ err_unlock:
   release_table_share(share);
 err_unlock2:
   pthread_mutex_unlock(&LOCK_open);
-  mdl_release_lock(&thd->mdl_context, mdl_lock_data);
+  if (! (flags & MYSQL_OPEN_HAS_MDL_LOCK))
+  {
+    mdl_release_lock(&thd->mdl_context, mdl_lock_data);
+    mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
+  }
   DBUG_RETURN(TRUE);
 }
 
@@ -3497,7 +3506,10 @@ recover_from_failed_open_table_attempt(THD *thd, TABLE_LIST *table,
       mdl_set_lock_type(table->mdl_lock_data, MDL_EXCLUSIVE);
       mdl_add_lock(&thd->mdl_context, table->mdl_lock_data);
       if (mdl_acquire_exclusive_locks(&thd->mdl_context))
+      {
+        mdl_remove_lock(&thd->mdl_context, table->mdl_lock_data);
         return TRUE;
+      }
       pthread_mutex_lock(&LOCK_open);
       tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name);
       ha_create_table_from_engine(thd, table->db, table->table_name);
@@ -3506,18 +3518,23 @@ recover_from_failed_open_table_attempt(THD *thd, TABLE_LIST *table,
       thd->warning_info->clear_warning_info(thd->query_id);
       thd->clear_error();                 // Clear error message
       mdl_release_lock(&thd->mdl_context, table->mdl_lock_data);
+      mdl_remove_lock(&thd->mdl_context, table->mdl_lock_data);
       break;
     case OT_REPAIR:
       mdl_set_lock_type(table->mdl_lock_data, MDL_EXCLUSIVE);
       mdl_add_lock(&thd->mdl_context, table->mdl_lock_data);
       if (mdl_acquire_exclusive_locks(&thd->mdl_context))
+      {
+        mdl_remove_lock(&thd->mdl_context, table->mdl_lock_data);
         return TRUE;
+      }
       pthread_mutex_lock(&LOCK_open);
       tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name);
       pthread_mutex_unlock(&LOCK_open);
 
       result= auto_repair_table(thd, table);
       mdl_release_lock(&thd->mdl_context, table->mdl_lock_data);
+      mdl_remove_lock(&thd->mdl_context, table->mdl_lock_data);
       break;
     default:
       DBUG_ASSERT(0);

sql/sql_delete.cc

@@ -1170,7 +1170,10 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
     mdl_set_lock_type(mdl_lock_data, MDL_EXCLUSIVE);
     mdl_add_lock(&thd->mdl_context, mdl_lock_data);
     if (mdl_acquire_exclusive_locks(&thd->mdl_context))
+    {
+      mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
       DBUG_RETURN(TRUE);
+    }
     pthread_mutex_lock(&LOCK_open);
     tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_list->db,
                      table_list->table_name);
@@ -1200,12 +1203,18 @@ end:
       my_ok(thd);		// This should return record count
     }
     if (mdl_lock_data)
+    {
       mdl_release_lock(&thd->mdl_context, mdl_lock_data);
+      mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
+    }
   }
   else if (error)
   {
     if (mdl_lock_data)
+    {
       mdl_release_lock(&thd->mdl_context, mdl_lock_data);
+      mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
+    }
   }
   DBUG_RETURN(error);
 

sql/sql_handler.cc

@@ -150,6 +150,7 @@ static void mysql_ha_close_table(THD *thd, TABLE_LIST *tables)
     }
     pthread_mutex_unlock(&LOCK_open);
     mdl_release_lock(&thd->handler_mdl_context, mdl_lock_data);
+    mdl_remove_lock(&thd->handler_mdl_context, mdl_lock_data);
   }
   else if (tables->table)
   {

sql/sql_show.cc

@@ -3250,6 +3250,7 @@ err_unlock:
 
 err:
   mdl_release_lock(&thd->mdl_context, &mdl_lock_data);
+  mdl_remove_lock(&thd->mdl_context, &mdl_lock_data);
   thd->clear_error();
   return res;
 }

sql/sql_table.cc

@@ -2208,7 +2208,8 @@ err:
           and locked and therefore have to remove several metadata lock
           requests associated with them.
         */
-        mdl_release_all_locks_for_name(&thd->mdl_context, table->mdl_lock_data);
+        mdl_release_and_remove_all_locks_for_name(&thd->mdl_context,
+                                                  table->mdl_lock_data);
       }
     }
   }
@@ -4080,13 +4081,27 @@ static bool lock_table_name_if_not_cached(THD *thd, const char *db,
                                           const char *table_name,
                                           MDL_LOCK_DATA **lock_data)
 {
+  bool conflict;
+
   if (!(*lock_data= mdl_alloc_lock(0, db, table_name, thd->mem_root)))
     return TRUE;
   mdl_set_lock_type(*lock_data, MDL_EXCLUSIVE);
   mdl_add_lock(&thd->mdl_context, *lock_data);
-  if (mdl_try_acquire_exclusive_lock(&thd->mdl_context, *lock_data))
+  if (mdl_try_acquire_exclusive_lock(&thd->mdl_context, *lock_data,
+                                     &conflict))
   {
-    *lock_data= 0;
+    /*
+      To simplify our life under LOCK TABLES we remove unsatisfied
+      lock request from the context.
+    */
+    mdl_remove_lock(&thd->mdl_context, *lock_data);
+    if (!conflict)
+    {
+      /* Probably OOM. */
+      return TRUE;
+    }
+    else
+      *lock_data= 0;
   }
   return FALSE;
 }
@@ -4157,7 +4172,10 @@ bool mysql_create_table(THD *thd, const char *db, const char *table_name,
 
 unlock:
   if (target_lock_data)
+  {
     mdl_release_lock(&thd->mdl_context, target_lock_data);
+    mdl_remove_lock(&thd->mdl_context, target_lock_data);
+  }
   pthread_mutex_lock(&LOCK_lock_db);
   if (!--creating_table && creating_database)
     pthread_cond_signal(&COND_refresh);
@@ -4416,7 +4434,10 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
     mdl_set_lock_type(mdl_lock_data, MDL_EXCLUSIVE);
     mdl_add_lock(&thd->mdl_context, mdl_lock_data);
     if (mdl_acquire_exclusive_locks(&thd->mdl_context))
+    {
+      mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
       DBUG_RETURN(0);
+    }
 
     pthread_mutex_lock(&LOCK_open);
     if (!(share= (get_table_share(thd, table_list, key, key_length, 0,
@@ -4550,7 +4571,10 @@ end:
   }
   /* In case of a temporary table there will be no metadata lock. */
   if (error && mdl_lock_data)
+  {
     mdl_release_lock(&thd->mdl_context, mdl_lock_data);
+    mdl_remove_lock(&thd->mdl_context, mdl_lock_data);
+  }
   DBUG_RETURN(error);
 }
 
@@ -5518,7 +5542,10 @@ binlog:
 
 err:
   if (target_lock_data)
+  {
     mdl_release_lock(&thd->mdl_context, target_lock_data);
+    mdl_remove_lock(&thd->mdl_context, target_lock_data);
+  }
   DBUG_RETURN(res);
 }
 
@@ -6876,7 +6903,9 @@ view_err:
       if (new_name != table_name || new_db != db)
       {
         mdl_release_lock(&thd->mdl_context, target_lock_data);
-        mdl_release_all_locks_for_name(&thd->mdl_context, mdl_lock_data);
+        mdl_remove_lock(&thd->mdl_context, target_lock_data);
+        mdl_release_and_remove_all_locks_for_name(&thd->mdl_context,
+                                                  mdl_lock_data);
       }
       else
         mdl_downgrade_exclusive_lock(&thd->mdl_context, mdl_lock_data);
@@ -7554,7 +7583,9 @@ view_err:
     if ((new_name != table_name || new_db != db))
     {
       mdl_release_lock(&thd->mdl_context, target_lock_data);
-      mdl_release_all_locks_for_name(&thd->mdl_context, mdl_lock_data);
+      mdl_remove_lock(&thd->mdl_context, target_lock_data);
+      mdl_release_and_remove_all_locks_for_name(&thd->mdl_context,
+                                                mdl_lock_data);
     }
     else
       mdl_downgrade_exclusive_lock(&thd->mdl_context, mdl_lock_data);
@@ -7613,7 +7644,10 @@ err:
     thd->abort_on_warning= save_abort_on_warning;
   }
   if (target_lock_data)
+  {
     mdl_release_lock(&thd->mdl_context, target_lock_data);
+    mdl_remove_lock(&thd->mdl_context, target_lock_data);
+  }
   DBUG_RETURN(TRUE);
 
 err_with_mdl:
@@ -7625,8 +7659,11 @@ err_with_mdl:
   */
   thd->locked_tables_list.unlink_all_closed_tables();
   if (target_lock_data)
+  {
     mdl_release_lock(&thd->mdl_context, target_lock_data);
-  mdl_release_all_locks_for_name(&thd->mdl_context, mdl_lock_data);
+    mdl_remove_lock(&thd->mdl_context, target_lock_data);
+  }
+  mdl_release_and_remove_all_locks_for_name(&thd->mdl_context, mdl_lock_data);
   DBUG_RETURN(TRUE);
 }
 /* mysql_alter_table */