MDEV-3917 multiple use locks (GET_LOCK) in one connection.

    The patch contributed by Konstantin Osipov applied.
    Native comments:
      Implement multiple user-level locks per connection.

      GET_LOCK() function in MySQL allows a connection  to hold at most
      one user level lock. Taking a new lock automatically releases the
      old lock, if any.

      The limit of one lock per session existed since  early versions
      of MySQL didn't have a deadlock detector for SQL locks.
      MDL patches in MySQL 5.5 added a deadlock detector,
      so starting from 5.5 it became possible to take multiple locks
      in any order -- a deadlock, should it occur, would be detected
      and an error returned to the client which closed the wait chain.

      This is exactly what is done in this patch: ULLs are moved
      to use MDL subsystem.

10 files changed, 463 insertions, 346 deletions

diff --git a/sql/item.cc b/sql/item.cc
index fd6a11c319b..90a652e92c7 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -198,7 +198,7 @@ Hybrid_type_traits_integer::fix_length_and_dec(Item *item, Item *arg) const
 
 void item_init(void)
 {
-  item_user_lock_init();
+  item_func_sleep_init();
   uuid_short_init();
 }
 
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 5316f89e730..31bf11284b5 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -3750,120 +3750,6 @@ udf_handler::~udf_handler()
 bool udf_handler::get_arguments() { return 0; }
 #endif /* HAVE_DLOPEN */
 
-/*
-** User level locks
-*/
-
-mysql_mutex_t LOCK_user_locks;
-static HASH hash_user_locks;
-
-class User_level_lock
-{
-  uchar *key;
-  size_t key_length;
-
-public:
-  int count;
-  bool locked;
-  mysql_cond_t cond;
-  my_thread_id thread_id;
-  void set_thread(THD *thd) { thread_id= thd->thread_id; }
-
-  User_level_lock(const uchar *key_arg,uint length, ulong id) 
-    :key_length(length),count(1),locked(1), thread_id(id)
-  {
-    key= (uchar*) my_memdup(key_arg,length,MYF(0));
-    mysql_cond_init(key_user_level_lock_cond, &cond, NULL);
-    if (key)
-    {
-      if (my_hash_insert(&hash_user_locks,(uchar*) this))
-      {
-	my_free(key);
-	key=0;
-      }
-    }
-  }
-  ~User_level_lock()
-  {
-    if (key)
-    {
-      my_hash_delete(&hash_user_locks,(uchar*) this);
-      my_free(key);
-    }
-    mysql_cond_destroy(&cond);
-  }
-  inline bool initialized() { return key != 0; }
-  friend void item_user_lock_release(User_level_lock *ull);
-  friend uchar *ull_get_key(const User_level_lock *ull, size_t *length,
-                            my_bool not_used);
-};
-
-uchar *ull_get_key(const User_level_lock *ull, size_t *length,
-                   my_bool not_used __attribute__((unused)))
-{
-  *length= ull->key_length;
-  return ull->key;
-}
-
-#ifdef HAVE_PSI_INTERFACE
-static PSI_mutex_key key_LOCK_user_locks;
-
-static PSI_mutex_info all_user_mutexes[]=
-{
-  { &key_LOCK_user_locks, "LOCK_user_locks", PSI_FLAG_GLOBAL}
-};
-
-static void init_user_lock_psi_keys(void)
-{
-  const char* category= "sql";
-  int count;
-
-  if (PSI_server == NULL)
-    return;
-
-  count= array_elements(all_user_mutexes);
-  PSI_server->register_mutex(category, all_user_mutexes, count);
-}
-#endif
-
-static bool item_user_lock_inited= 0;
-
-void item_user_lock_init(void)
-{
-#ifdef HAVE_PSI_INTERFACE
-  init_user_lock_psi_keys();
-#endif
-
-  mysql_mutex_init(key_LOCK_user_locks, &LOCK_user_locks, MY_MUTEX_INIT_SLOW);
-  my_hash_init(&hash_user_locks,system_charset_info,
-	    16,0,0,(my_hash_get_key) ull_get_key,NULL,0);
-  item_user_lock_inited= 1;
-}
-
-void item_user_lock_free(void)
-{
-  if (item_user_lock_inited)
-  {
-    item_user_lock_inited= 0;
-    my_hash_free(&hash_user_locks);
-    mysql_mutex_destroy(&LOCK_user_locks);
-  }
-}
-
-void item_user_lock_release(User_level_lock *ull)
-{
-  ull->locked=0;
-  ull->thread_id= 0;
-  if (--ull->count)
-    mysql_cond_signal(&ull->cond);
-  else
-    delete ull;
-}
-
-/**
-  Wait until we are at or past the given position in the master binlog
-  on the slave.
-*/
 
 longlong Item_master_pos_wait::val_int()
 {
@@ -4010,7 +3896,136 @@ int Interruptible_wait::wait(mysql_cond_t *cond, mysql_mutex_t *mutex)
 
 
 /**
-  Get a user level lock.  If the thread has an old lock this is first released.
+  For locks with EXPLICIT duration, MDL returns a new ticket
+  every time a lock is granted. This allows to implement recursive
+  locks without extra allocation or additional data structures, such
+  as below. However, if there are too many tickets in the same
+  MDL_context, MDL_context::find_ticket() is getting too slow,
+  since it's using a linear search.
+  This is why a separate structure is allocated for a user
+  level lock, and before requesting a new lock from MDL,
+  GET_LOCK() checks thd->ull_hash if such lock is already granted,
+  and if so, simply increments a reference counter.
+*/
+
+class User_level_lock
+{
+public:
+  MDL_ticket *lock;
+  int refs;
+};
+
+
+/** Extract a hash key from User_level_lock. */
+
+uchar *ull_get_key(const uchar *ptr, size_t *length,
+                   my_bool not_used __attribute__((unused)))
+{
+  User_level_lock *ull = (User_level_lock*) ptr;
+  MDL_key *key = ull->lock->get_key();
+  *length= key->length();
+  return (uchar*) key->ptr();
+}
+
+
+/**
+  Release all user level locks for this THD.
+*/
+
+void mysql_ull_cleanup(THD *thd)
+{
+  User_level_lock *ull;
+  DBUG_ENTER("mysql_ull_cleanup");
+
+  for (uint i= 0; i < thd->ull_hash.records; i++)
+  {
+    ull = (User_level_lock*) my_hash_element(&thd->ull_hash, i);
+    thd->mdl_context.release_lock(ull->lock);
+    my_free(ull);
+  }
+
+  my_hash_free(&thd->ull_hash);
+
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+  Set explicit duration for metadata locks corresponding to
+  user level locks to protect them from being released at the end
+  of transaction.
+*/
+
+void mysql_ull_set_explicit_lock_duration(THD *thd)
+{
+  User_level_lock *ull;
+  DBUG_ENTER("mysql_ull_set_explicit_lock_duration");
+
+  for (uint i= 0; i < thd->ull_hash.records; i++)
+  {
+    ull= (User_level_lock*) my_hash_element(&thd->ull_hash, i);
+    thd->mdl_context.set_lock_duration(ull->lock, MDL_EXPLICIT);
+  }
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+  When MDL detects a lock wait timeout, it pushes
+  an error into the statement diagnostics area.
+  For GET_LOCK(), lock wait timeout is not an error,
+  but a special return value (0). NULL is returned in
+  case of error.
+  Capture and suppress lock wait timeout.
+*/
+
+class Lock_wait_timeout_handler: public Internal_error_handler
+{
+public:
+  Lock_wait_timeout_handler() :m_lock_wait_timeout(false) {}
+
+  bool m_lock_wait_timeout;
+
+  bool handle_condition(THD * /* thd */, uint sql_errno,
+                        const char * /* sqlstate */,
+                        MYSQL_ERROR::enum_warning_level /* level */,
+                        const char *message,
+                        MYSQL_ERROR ** /* cond_hdl */);
+};
+
+bool
+Lock_wait_timeout_handler::
+handle_condition(THD * /* thd */, uint sql_errno,
+                 const char * /* sqlstate */,
+                 MYSQL_ERROR::enum_warning_level /* level */,
+                 const char *message,
+                 MYSQL_ERROR ** /* cond_hdl */)
+{
+  if (sql_errno == ER_LOCK_WAIT_TIMEOUT)
+  {
+    m_lock_wait_timeout= true;
+    return true;                                /* condition handled */
+  }
+  return false;
+}
+
+
+static int ull_name_ok(String *name)
+{
+  if (!name || !name->length())
+    return 0;
+
+  if (name->length() > NAME_LEN)
+  {
+    my_error(ER_TOO_LONG_IDENT, MYF(0), name->c_ptr_safe());
+    return 0;
+  }
+  return 1;
+}
+
+
+/**
+  Get a user level lock.
 
   @retval
     1    : Got lock
@@ -4023,14 +4038,13 @@ int Interruptible_wait::wait(mysql_cond_t *cond, mysql_mutex_t *mutex)
 longlong Item_func_get_lock::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  String *res=args[0]->val_str(&value);
+  String *res= args[0]->val_str(&value);
   ulonglong timeout= args[1]->val_int();
-  THD *thd=current_thd;
+  THD *thd= current_thd;
   User_level_lock *ull;
-  int error;
-  Interruptible_wait timed_cond(thd);
   DBUG_ENTER("Item_func_get_lock::val_int");
 
+  null_value= 1;
   /*
     In slave thread no need to get locks, everything is serialized. Anyway
     there is no way to make GET_LOCK() work on slave like it did on master
@@ -4039,104 +4053,70 @@ longlong Item_func_get_lock::val_int()
     it's not guaranteed to be same as on master.
   */
   if (thd->slave_thread)
+  {
+    null_value= 0;
     DBUG_RETURN(1);
+  }
 
-  mysql_mutex_lock(&LOCK_user_locks);
-
-  if (!res || !res->length())
-  {
-    mysql_mutex_unlock(&LOCK_user_locks);
-    null_value=1;
+  if (!ull_name_ok(res))
     DBUG_RETURN(0);
-  }
+
   DBUG_PRINT("info", ("lock %.*s, thd=%ld", res->length(), res->ptr(),
                       (long) thd->real_id));
-  null_value=0;
-
-  if (thd->ull)
+  /* HASH entries are of type User_level_lock. */
+  if (! my_hash_inited(&thd->ull_hash) &&
+        my_hash_init(&thd->ull_hash, &my_charset_bin,
+                     16 /* small hash */, 0, 0, ull_get_key, NULL, 0))
   {
-    item_user_lock_release(thd->ull);
-    thd->ull=0;
-  }
-
-  if (!(ull= ((User_level_lock *) my_hash_search(&hash_user_locks,
-                                                 (uchar*) res->ptr(),
-                                                 (size_t) res->length()))))
-  {
-    ull= new User_level_lock((uchar*) res->ptr(), (size_t) res->length(),
-                             thd->thread_id);
-    if (!ull || !ull->initialized())
-    {
-      delete ull;
-      mysql_mutex_unlock(&LOCK_user_locks);
-      null_value=1;				// Probably out of memory
-      DBUG_RETURN(0);
-    }
-    ull->set_thread(thd);
-    thd->ull=ull;
-    mysql_mutex_unlock(&LOCK_user_locks);
-    DBUG_PRINT("info", ("made new lock"));
-    DBUG_RETURN(1);				// Got new lock
+    DBUG_RETURN(0);
   }
-  ull->count++;
-  DBUG_PRINT("info", ("ull->count=%d", ull->count));
 
-  /*
-    Structure is now initialized.  Try to get the lock.
-    Set up control struct to allow others to abort locks.
-  */
-  thd_proc_info(thd, "User lock");
-  thd->mysys_var->current_mutex= &LOCK_user_locks;
-  thd->mysys_var->current_cond=  &ull->cond;
+  MDL_request ull_request;
+  ull_request.init(MDL_key::USER_LOCK, res->c_ptr_safe(), "",
+                   MDL_SHARED_NO_WRITE, MDL_EXPLICIT);
+  MDL_key *ull_key = &ull_request.key;
 
-  timed_cond.set_timeout(timeout * 1000000000ULL);
 
-  error= 0;
-  thd_wait_begin(thd, THD_WAIT_USER_LOCK);
-  while (ull->locked && !thd->killed)
+  if ((ull= (User_level_lock*)
+       my_hash_search(&thd->ull_hash, ull_key->ptr(), ull_key->length())))
   {
-    DBUG_PRINT("info", ("waiting on lock"));
-    error= timed_cond.wait(&ull->cond, &LOCK_user_locks);
-    if (error == ETIMEDOUT || error == ETIME)
-    {
-      DBUG_PRINT("info", ("lock wait timeout"));
-      break;
-    }
-    error= 0;
+    /* Recursive lock */
+    ull->refs++;
+    null_value = 0;
+    DBUG_RETURN(1);
   }
-  thd_wait_end(thd);
 
-  if (ull->locked)
+  Lock_wait_timeout_handler lock_wait_timeout_handler;
+  thd->push_internal_handler(&lock_wait_timeout_handler);
+  bool error= thd->mdl_context.acquire_lock(&ull_request, timeout);
+  (void) thd->pop_internal_handler();
+  if (error)
   {
-    if (!--ull->count)
-    {
-      DBUG_ASSERT(0);
-      delete ull;				// Should never happen
-    }
-    if (!error)                                 // Killed (thd->killed != 0)
-    {
-      error=1;
-      null_value=1;				// Return NULL
-    }
+    if (lock_wait_timeout_handler.m_lock_wait_timeout)
+      null_value= 0;
+    DBUG_RETURN(0);
   }
-  else                                          // We got the lock
+
+  ull= (User_level_lock*) my_malloc(sizeof(User_level_lock),
+                                    MYF(MY_WME|MY_THREAD_SPECIFIC));
+  if (ull == NULL)
   {
-    ull->locked=1;
-    ull->set_thread(thd);
-    ull->thread_id= thd->thread_id;
-    thd->ull=ull;
-    error=0;
-    DBUG_PRINT("info", ("got the lock"));
+    thd->mdl_context.release_lock(ull_request.ticket);
+    DBUG_RETURN(0);
   }
-  mysql_mutex_unlock(&LOCK_user_locks);
 
-  mysql_mutex_lock(&thd->mysys_var->mutex);
-  thd_proc_info(thd, 0);
-  thd->mysys_var->current_mutex= 0;
-  thd->mysys_var->current_cond=  0;
-  mysql_mutex_unlock(&thd->mysys_var->mutex);
+  ull->lock= ull_request.ticket;
+  ull->refs= 1;
 
-  DBUG_RETURN(!error ? 1 : 0);
+  if (my_hash_insert(&thd->ull_hash, (uchar*) ull))
+  {
+    thd->mdl_context.release_lock(ull->lock);
+    my_free(ull);
+    DBUG_RETURN(0);
+  }
+  null_value= 0;
+
+  DBUG_RETURN(1);
 }
 
 
@@ -4151,43 +4131,86 @@ longlong Item_func_get_lock::val_int()
 longlong Item_func_release_lock::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  String *res=args[0]->val_str(&value);
-  User_level_lock *ull;
-  longlong result;
-  THD *thd=current_thd;
+  String *res= args[0]->val_str(&value);
+  THD *thd= current_thd;
   DBUG_ENTER("Item_func_release_lock::val_int");
-  if (!res || !res->length())
-  {
-    null_value=1;
+  null_value= 1;
+
+  if (!ull_name_ok(res))
     DBUG_RETURN(0);
-  }
+
   DBUG_PRINT("info", ("lock %.*s", res->length(), res->ptr()));
-  null_value=0;
 
-  result=0;
-  mysql_mutex_lock(&LOCK_user_locks);
-  if (!(ull= ((User_level_lock*) my_hash_search(&hash_user_locks,
-                                                (const uchar*) res->ptr(),
-                                                (size_t) res->length()))))
+  MDL_key ull_key;
+  ull_key.mdl_key_init(MDL_key::USER_LOCK, res->c_ptr_safe(), "");
+
+  User_level_lock *ull;
+
+  if (!(ull=
+        (User_level_lock*) my_hash_search(&thd->ull_hash,
+                                          ull_key.ptr(), ull_key.length())))
   {
-    null_value=1;
+    null_value= thd->mdl_context.get_lock_owner(&ull_key) == 0;
+    DBUG_RETURN(0);
   }
-  else
+  null_value= 0;
+  if (--ull->refs == 0)
   {
-    DBUG_PRINT("info", ("ull->locked=%d ull->thread=%lu thd=%lu", 
-                        (int) ull->locked,
-                        (long)ull->thread_id,
-                        (long)thd->thread_id));
-    if (ull->locked && current_thd->thread_id == ull->thread_id)
-    {
-      DBUG_PRINT("info", ("release lock"));
-      result=1;					// Release is ok
-      item_user_lock_release(ull);
-      thd->ull=0;
-    }
+    my_hash_delete(&thd->ull_hash, (uchar*) ull);
+    thd->mdl_context.release_lock(ull->lock);
+    my_free(ull);
   }
-  mysql_mutex_unlock(&LOCK_user_locks);
-  DBUG_RETURN(result);
+  DBUG_RETURN(1);
+}
+
+
+/**
+  Check a user level lock.
+
+  Sets null_value=TRUE on error.
+
+  @retval
+    1		Available
+  @retval
+    0		Already taken, or error
+*/
+
+longlong Item_func_is_free_lock::val_int()
+{
+  DBUG_ASSERT(fixed == 1);
+  String *res= args[0]->val_str(&value);
+  THD *thd= current_thd;
+  null_value= 1;
+
+  if (!ull_name_ok(res))
+    return 0;
+
+  MDL_key ull_key;
+  ull_key.mdl_key_init(MDL_key::USER_LOCK, res->c_ptr_safe(), "");
+
+  null_value= 0;
+  return thd->mdl_context.get_lock_owner(&ull_key) == 0;
+}
+
+
+longlong Item_func_is_used_lock::val_int()
+{
+  DBUG_ASSERT(fixed == 1);
+  String *res= args[0]->val_str(&value);
+  THD *thd= current_thd;
+  null_value= 1;
+
+  if (!ull_name_ok(res))
+    return 0;
+
+  MDL_key ull_key;
+  ull_key.mdl_key_init(MDL_key::USER_LOCK, res->c_ptr_safe(), "");
+  ulong thread_id = thd->mdl_context.get_lock_owner(&ull_key);
+  if (thread_id == 0)
+    return 0;
+
+  null_value= 0;
+  return thread_id;
 }
 
 
@@ -4288,6 +4311,54 @@ void Item_func_benchmark::print(String *str, enum_query_type query_type)
 }
 
 
+mysql_mutex_t LOCK_item_func_sleep;
+
+#ifdef HAVE_PSI_INTERFACE
+static PSI_mutex_key key_LOCK_item_func_sleep;
+
+static PSI_mutex_info item_func_sleep_mutexes[]=
+{
+  { &key_LOCK_item_func_sleep, "LOCK_user_locks", PSI_FLAG_GLOBAL}
+};
+
+
+static void init_item_func_sleep_psi_keys(void)
+{
+  const char* category= "sql";
+  int count;
+
+  if (PSI_server == NULL)
+    return;
+
+  count= array_elements(item_func_sleep_mutexes);
+  PSI_server->register_mutex(category, item_func_sleep_mutexes, count);
+}
+#endif
+
+static bool item_func_sleep_inited= 0;
+
+
+void item_func_sleep_init(void)
+{
+#ifdef HAVE_PSI_INTERFACE
+  init_item_func_sleep_psi_keys();
+#endif
+
+  mysql_mutex_init(key_LOCK_item_func_sleep, &LOCK_item_func_sleep, MY_MUTEX_INIT_SLOW);
+  item_func_sleep_inited= 1;
+}
+
+
+void item_func_sleep_free(void)
+{
+  if (item_func_sleep_inited)
+  {
+    item_func_sleep_inited= 0;
+    mysql_mutex_destroy(&LOCK_item_func_sleep);
+  }
+}
+
+
 /** This function is just used to create tests with time gaps. */
 
 longlong Item_func_sleep::val_int()
@@ -4316,24 +4387,24 @@ longlong Item_func_sleep::val_int()
   timed_cond.set_timeout((ulonglong) (timeout * 1000000000.0));
 
   mysql_cond_init(key_item_func_sleep_cond, &cond, NULL);
-  mysql_mutex_lock(&LOCK_user_locks);
+  mysql_mutex_lock(&LOCK_item_func_sleep);
 
   thd_proc_info(thd, "User sleep");
-  thd->mysys_var->current_mutex= &LOCK_user_locks;
+  thd->mysys_var->current_mutex= &LOCK_item_func_sleep;
   thd->mysys_var->current_cond=  &cond;
 
   error= 0;
   thd_wait_begin(thd, THD_WAIT_SLEEP);
   while (!thd->killed)
   {
-    error= timed_cond.wait(&cond, &LOCK_user_locks);
+    error= timed_cond.wait(&cond, &LOCK_item_func_sleep);
     if (error == ETIMEDOUT || error == ETIME)
       break;
     error= 0;
   }
   thd_wait_end(thd);
   thd_proc_info(thd, 0);
-  mysql_mutex_unlock(&LOCK_user_locks);
+  mysql_mutex_unlock(&LOCK_item_func_sleep);
   mysql_mutex_lock(&thd->mysys_var->mutex);
   thd->mysys_var->current_mutex= 0;
   thd->mysys_var->current_cond=  0;
@@ -6208,61 +6279,6 @@ Item *get_system_var(THD *thd, enum_var_type var_type, LEX_STRING name,
 }
 
 
-/**
-  Check a user level lock.
-
-  Sets null_value=TRUE on error.
-
-  @retval
-    1		Available
-  @retval
-    0		Already taken, or error
-*/
-
-longlong Item_func_is_free_lock::val_int()
-{
-  DBUG_ASSERT(fixed == 1);
-  String *res=args[0]->val_str(&value);
-  User_level_lock *ull;
-
-  null_value=0;
-  if (!res || !res->length())
-  {
-    null_value=1;
-    return 0;
-  }
-  
-  mysql_mutex_lock(&LOCK_user_locks);
-  ull= (User_level_lock *) my_hash_search(&hash_user_locks, (uchar*) res->ptr(),
-                                          (size_t) res->length());
-  mysql_mutex_unlock(&LOCK_user_locks);
-  if (!ull || !ull->locked)
-    return 1;
-  return 0;
-}
-
-longlong Item_func_is_used_lock::val_int()
-{
-  DBUG_ASSERT(fixed == 1);
-  String *res=args[0]->val_str(&value);
-  User_level_lock *ull;
-
-  null_value=1;
-  if (!res || !res->length())
-    return 0;
-  
-  mysql_mutex_lock(&LOCK_user_locks);
-  ull= (User_level_lock *) my_hash_search(&hash_user_locks, (uchar*) res->ptr(),
-                                          (size_t) res->length());
-  mysql_mutex_unlock(&LOCK_user_locks);
-  if (!ull || !ull->locked)
-    return 0;
-
-  null_value=0;
-  return ull->thread_id;
-}
-
-
 longlong Item_func_row_count::val_int()
 {
   DBUG_ASSERT(fixed == 1);
diff --git a/sql/item_func.h b/sql/item_func.h
index 653641c9f72..ab6ec706248 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -1257,6 +1257,9 @@ public:
 };
 
 
+void item_func_sleep_init(void);
+void item_func_sleep_free(void);
+
 class Item_func_sleep :public Item_int_func
 {
 public:
@@ -1506,14 +1509,8 @@ public:
 
 #endif /* HAVE_DLOPEN */
 
-/*
-** User level locks
-*/
-
-class User_level_lock;
-void item_user_lock_init(void);
-void item_user_lock_release(User_level_lock *ull);
-void item_user_lock_free(void);
+void mysql_ull_cleanup(THD *thd);
+void mysql_ull_set_explicit_lock_duration(THD *thd);
 
 class Item_func_get_lock :public Item_int_func
 {
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 7fd522d053a..e739a9aff78 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -85,7 +85,8 @@ const char *MDL_key::m_namespace_to_wait_state_name[NAMESPACE_END]=
   "Waiting for stored procedure metadata lock",
   "Waiting for trigger metadata lock",
   "Waiting for event metadata lock",
-  "Waiting for commit lock"
+  "Waiting for commit lock",
+  "User lock" /* Be compatible with old status. */
 };
 
 static bool mdl_initialized= 0;
@@ -107,6 +108,7 @@ public:
   void init();
   void destroy();
   MDL_lock *find_or_insert(const MDL_key *key);
+  unsigned long get_lock_owner(const MDL_key *key);
   void remove(MDL_lock *lock);
 private:
   bool move_from_hash_to_lock_mutex(MDL_lock *lock);
@@ -382,6 +384,7 @@ public:
                       bool ignore_lock_priority) const;
 
   inline static MDL_lock *create(const MDL_key *key);
+  inline unsigned long get_lock_owner() const;
 
   void reschedule_waiters();
 
@@ -857,6 +860,43 @@ bool MDL_map::move_from_hash_to_lock_mutex(MDL_lock *lock)
 
 
 /**
+ * Return thread id of the owner of the lock, if it is owned.
+ */
+
+unsigned long
+MDL_map::get_lock_owner(const MDL_key *mdl_key)
+{
+  MDL_lock *lock;
+  unsigned long res= 0;
+
+  if (mdl_key->mdl_namespace() == MDL_key::GLOBAL ||
+      mdl_key->mdl_namespace() == MDL_key::COMMIT)
+  {
+    lock= (mdl_key->mdl_namespace() == MDL_key::GLOBAL) ? m_global_lock :
+                                                          m_commit_lock;
+    mysql_prlock_rdlock(&lock->m_rwlock);
+    res= lock->get_lock_owner();
+    mysql_prlock_unlock(&lock->m_rwlock);
+  }
+  else
+  {
+    my_hash_value_type hash_value= my_calc_hash(&m_locks,
+                                                mdl_key->ptr(),
+                                                mdl_key->length());
+    mysql_mutex_lock(&m_mutex);
+    lock= (MDL_lock*) my_hash_search_using_hash_value(&m_locks,
+                                                    hash_value,
+                                                    mdl_key->ptr(),
+                                                    mdl_key->length());
+    if (lock)
+      res= lock->get_lock_owner();
+    mysql_mutex_unlock(&m_mutex);
+  }
+  return res;
+}
+
+
+/**
   Destroy MDL_lock object or delegate this responsibility to
   whatever thread that holds the last outstanding reference to
   it.
@@ -1621,6 +1661,23 @@ MDL_lock::can_grant_lock(enum_mdl_type type_arg,
 }
 
 
+/**
+  Return thread id of the thread to which the first ticket was
+  granted.
+*/
+
+inline unsigned long
+MDL_lock::get_lock_owner() const
+{
+  Ticket_iterator it(m_granted);
+  MDL_ticket *ticket;
+
+  if ((ticket= it++))
+    return thd_get_thread_id(ticket->get_ctx()->get_thd());
+  return 0;
+}
+
+
 /** Remove a ticket from waiting or pending queue and wakeup up waiters. */
 
 void MDL_lock::remove_ticket(Ticket_list MDL_lock::*list, MDL_ticket *ticket)
@@ -2094,31 +2151,37 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
 
   find_deadlock();
 
-  if (lock->needs_notification(ticket))
+  struct timespec abs_shortwait;
+  set_timespec(abs_shortwait, 1);
+  wait_status= MDL_wait::EMPTY;
+
+  while (cmp_timespec(abs_shortwait, abs_timeout) <= 0)
   {
-    struct timespec abs_shortwait;
-    set_timespec(abs_shortwait, 1);
-    wait_status= MDL_wait::EMPTY;
+    /* abs_timeout is far away. Wait a short while and notify locks. */
+    wait_status= m_wait.timed_wait(m_thd, &abs_shortwait, FALSE,
+                                   mdl_request->key.get_wait_state_name());
 
-    while (cmp_timespec(abs_shortwait, abs_timeout) <= 0)
+    if (wait_status != MDL_wait::EMPTY)
+      break;
+    /* Check if the client is gone while we were waiting. */
+    if (! thd_is_connected(m_thd))
     {
-      /* abs_timeout is far away. Wait a short while and notify locks. */
-      wait_status= m_wait.timed_wait(m_thd, &abs_shortwait, FALSE,
-                                     mdl_request->key.get_wait_state_name());
-
-      if (wait_status != MDL_wait::EMPTY)
-        break;
+      /*
+       * The client is disconnected. Don't wait forever:
+       * assume it's the same as a wait timeout, this
+       * ensures all error handling is correct.
+       */
+      wait_status= MDL_wait::TIMEOUT;
+      break;
+    }
 
-      mysql_prlock_wrlock(&lock->m_rwlock);
+    mysql_prlock_wrlock(&lock->m_rwlock);
+    if (lock->needs_notification(ticket))
       lock->notify_conflicting_locks(this);
-      mysql_prlock_unlock(&lock->m_rwlock);
-      set_timespec(abs_shortwait, 1);
-    }
-    if (wait_status == MDL_wait::EMPTY)
-      wait_status= m_wait.timed_wait(m_thd, &abs_timeout, TRUE,
-                                     mdl_request->key.get_wait_state_name());
+    mysql_prlock_unlock(&lock->m_rwlock);
+    set_timespec(abs_shortwait, 1);
   }
-  else
+  if (wait_status == MDL_wait::EMPTY)
     wait_status= m_wait.timed_wait(m_thd, &abs_timeout, TRUE,
                                    mdl_request->key.get_wait_state_name());
 
@@ -2613,7 +2676,7 @@ void MDL_context::release_lock(MDL_ticket *ticket)
   the corresponding lists, i.e. stored in reverse temporal order.
   This allows to employ this function to:
   - back off in case of a lock conflict.
-  - release all locks in the end of a statment or transaction
+  - release all locks in the end of a statement or transaction
   - rollback to a savepoint.
 */
 
@@ -2725,6 +2788,22 @@ MDL_context::is_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
 
 
 /**
+  Return thread id of the owner of the lock or 0 if
+  there is no owner.
+  @note: Lock type is not considered at all, the function
+  simply checks that there is some lock for the given key.
+
+  @return  thread id of the owner of the lock or 0
+*/
+
+unsigned long
+MDL_context::get_lock_owner(MDL_key *key)
+{
+  return mdl_locks.get_lock_owner(key);
+}
+
+
+/**
   Check if we have any pending locks which conflict with existing shared lock.
 
   @pre The ticket must match an acquired lock.
@@ -2737,6 +2816,11 @@ bool MDL_ticket::has_pending_conflicting_lock() const
   return m_lock->has_pending_conflicting_lock(m_type);
 }
 
+/** Return a key identifying this lock. */
+MDL_key *MDL_ticket::get_key() const
+{
+        return &m_lock->key;
+}
 
 /**
   Releases metadata locks that were acquired after a specific savepoint.
diff --git a/sql/mdl.h b/sql/mdl.h
index 68f24a7a0e8..a86b45e180f 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -212,6 +212,7 @@ public:
                             TRIGGER,
                             EVENT,
                             COMMIT,
+                            USER_LOCK,           /* user level locks. */
                             /* This should be the last ! */
                             NAMESPACE_END };
 
@@ -492,6 +493,7 @@ public:
   }
   enum_mdl_type get_type() const { return m_type; }
   MDL_lock *get_lock() const { return m_lock; }
+  MDL_key *get_key() const;
   void downgrade_exclusive_lock(enum_mdl_type type);
 
   bool has_stronger_or_equal_type(enum_mdl_type type) const;
@@ -653,6 +655,7 @@ public:
   bool is_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
                      const char *db, const char *name,
                      enum_mdl_type mdl_type);
+  unsigned long get_lock_owner(MDL_key *mdl_key);
 
   bool has_lock(const MDL_savepoint &mdl_savepoint, MDL_ticket *mdl_ticket);
 
@@ -721,9 +724,9 @@ private:
     Lists of MDL tickets:
     ---------------------
     The entire set of locks acquired by a connection can be separated
-    in three subsets according to their: locks released at the end of
-    statement, at the end of transaction and locks are released
-    explicitly.
+    in three subsets according to their duration: locks released at
+    the end of statement, at the end of transaction and locks are
+    released explicitly.
 
     Statement and transactional locks are locks with automatic scope.
     They are accumulated in the course of a transaction, and released
@@ -732,11 +735,12 @@ private:
     locks). They must not be (and never are) released manually,
     i.e. with release_lock() call.
 
-    Locks with explicit duration are taken for locks that span
+    Tickets with explicit duration 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
+    under LOCK TABLES, and the locked tables stay locked), user level
+    locks (GET_LOCK()/RELEASE_LOCK() functions) and
     locks implementing "global read lock".
 
     Statement/transactional locks are always prepended to the
@@ -745,20 +749,19 @@ private:
     a savepoint, we start popping and releasing tickets from the
     front until we reach the last ticket acquired after the savepoint.
 
-    Locks with explicit duration stored are not stored in any
+    Locks with explicit duration are not stored in any
     particular order, and among each other can be split into
-    three sets:
+    four sets:
 
-    [LOCK TABLES locks] [HANDLER locks] [GLOBAL READ LOCK locks]
+    [LOCK TABLES locks] [USER locks] [HANDLER locks] [GLOBAL READ LOCK locks]
 
     The following is known about these sets:
 
-    * 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
+    * GLOBAL READ LOCK locks are always stored last.
+      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
@@ -853,6 +856,18 @@ extern bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use,
 extern "C" const char* thd_enter_cond(MYSQL_THD thd, mysql_cond_t *cond,
                                       mysql_mutex_t *mutex, const char *msg);
 extern "C" void thd_exit_cond(MYSQL_THD thd, const char *old_msg);
+extern "C" unsigned long thd_get_thread_id(const MYSQL_THD thd);
+
+/**
+  Check if a connection in question is no longer connected.
+
+  @details
+  Replication apply thread is always connected. Otherwise,
+  does a poll on the associated socket to check if the client
+  is gone.
+*/
+extern "C" int thd_is_connected(MYSQL_THD thd);
+
 
 #ifndef DBUG_OFF
 extern mysql_mutex_t LOCK_open;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index c8d8c09cbba..3f569e31d08 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1824,7 +1824,7 @@ void clean_up(bool print_message)
 #endif
   query_cache_destroy();
   hostname_cache_free();
-  item_user_lock_free();
+  item_func_sleep_free();
   lex_free();				/* Free some memory */
   item_create_cleanup();
   if (!opt_noacl)
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 587673de37c..9f9f288b6d5 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -335,7 +335,7 @@ extern MYSQL_PLUGIN_IMPORT key_map key_map_full;          /* Should be threaded
   Server mutex locks and condition variables.
  */
 extern mysql_mutex_t
-       LOCK_user_locks, LOCK_status,
+       LOCK_item_func_sleep, LOCK_status,
        LOCK_error_log, LOCK_delayed_insert, LOCK_short_uuid_generator,
        LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
        LOCK_slave_list, LOCK_active_mi, LOCK_manager,
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 95aa4a00122..caf56f9f84c 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -827,6 +827,7 @@ THD::THD()
   col_access=0;
   is_slave_error= thread_specific_used= FALSE;
   my_hash_clear(&handler_tables_hash);
+  my_hash_clear(&ull_hash);
   tmp_table=0;
   cuted_fields= 0L;
   sent_row_count= 0L;
@@ -866,7 +867,6 @@ THD::THD()
   net.vio=0;
   net.buff= 0;
   client_capabilities= 0;                       // minimalistic client
-  ull=0;
   system_thread= NON_SYSTEM_THREAD;
   cleanup_done= abort_on_warning= 0;
   peer_port= 0;					// For SHOW PROCESSLIST
@@ -1400,8 +1400,6 @@ void THD::cleanup(void)
   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 (user_connect)
   {
     decrease_user_connections(user_connect);
@@ -1419,13 +1417,9 @@ void THD::cleanup(void)
   sp_cache_clear(&sp_proc_cache);
   sp_cache_clear(&sp_func_cache);
 
-  if (ull)
-  {
-    mysql_mutex_lock(&LOCK_user_locks);
-    item_user_lock_release(ull);
-    mysql_mutex_unlock(&LOCK_user_locks);
-    ull= NULL;
-  }
+  mysql_ull_cleanup(this);
+  /* All metadata locks must have been released by now. */
+  DBUG_ASSERT(!mdl_context.has_locks());
 
   apc_target.destroy();
   cleanup_done=1;
@@ -4001,6 +3995,15 @@ extern "C" unsigned long thd_get_thread_id(const MYSQL_THD thd)
 }
 
 
+/**
+  Check if THD socket is still connected.
+ */
+extern "C" int thd_is_connected(MYSQL_THD thd)
+{
+  return thd->is_connected();
+}
+
+
 #ifdef INNODB_COMPATIBILITY_HOOKS
 extern "C" const struct charset_info_st *thd_charset(MYSQL_THD thd)
 {
@@ -4321,6 +4324,8 @@ void THD::leave_locked_tables_mode()
     /* Also ensure that we don't release metadata locks for open HANDLERs. */
     if (handler_tables_hash.records)
       mysql_ha_set_explicit_lock_duration(this);
+    if (ull_hash.records)
+      mysql_ull_set_explicit_lock_duration(this);
   }
   locked_tables_mode= LTM_NONE;
 }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 8ccb524f715..6b779484bbc 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -57,7 +57,6 @@ class Lex_input_stream;
 class Parser_state;
 class Rows_log_event;
 class Sroutine_hash_entry;
-class User_level_lock;
 class user_var_entry;
 
 enum enum_enable_or_disable { LEAVE_AS_IS, ENABLE, DISABLE };
@@ -1682,11 +1681,11 @@ public:
 
   HASH		handler_tables_hash;
   /*
-    One thread can hold up to one named user-level lock. This variable
-    points to a lock object if the lock is present. See item_func.cc and
+    A thread can hold named user-level locks. This variable
+    contains granted tickets if a lock is present. See item_func.cc and
     chapter 'Miscellaneous functions', for functions GET_LOCK, RELEASE_LOCK.
   */
-  User_level_lock *ull;
+  HASH ull_hash;
 #ifndef DBUG_OFF
   uint dbug_sentry; // watch out for memory corruption
 #endif
diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc
index 2720dc7cd74..e9c9dc86e41 100644
--- a/sql/sql_reload.cc
+++ b/sql/sql_reload.cc
@@ -205,6 +205,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long options,
   DBUG_ASSERT(!thd || thd->locked_tables_mode ||
               !thd->mdl_context.has_locks() ||
               thd->handler_tables_hash.records ||
+              thd->ull_hash.records ||
               thd->global_read_lock.is_acquired());
 
   /*