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.

mysql-test/r/func_misc.result

@@ -338,6 +338,227 @@ set optimizer_switch=@optimizer_switch_save;
 drop view  v_merge, vm;
 drop table t1,tv;
 #
+# GET_LOCK, RELEASE_LOCK, IS_USED_LOCK functions test
+#
+# IS_USED_LOCK, IS_FREE_LOCK: the  lock is not acquired
+# Note: IS_USED_LOCK returns NULL if the lock is unused
+select is_used_lock('test');
+is_used_lock('test')
+NULL
+select is_free_lock('test');
+is_free_lock('test')
+1
+# GET_LOCK returns 1 if it manages to acquire a lock
+select get_lock('test', 0);
+get_lock('test', 0)
+1
+# IS_USED_LOCK, IS_FREE_LOCK: the  lock is acquired
+select is_free_lock('test');
+is_free_lock('test')
+0
+select is_used_lock('test') = connection_id();
+is_used_lock('test') = connection_id()
+1
+# -> Switching to connection 'con1'
+# IS_USED_LOCK, IS_FREE_LOCK: the  lock is acquired in another
+# connection
+select is_used_lock('test') = connection_id();
+is_used_lock('test') = connection_id()
+0
+select is_free_lock('test');
+is_free_lock('test')
+0
+# GET_LOCK returns 0 if it can't acquire a lock (wait timeout)
+select get_lock('test', 0);
+get_lock('test', 0)
+0
+# RELEASE_LOCK returns 0 if the lock belongs to another connection 
+select release_lock('test');
+release_lock('test')
+0
+# -> Switching to connection 'default'
+# RELEASE_LOCK returns 1 if it successfully releases a lock
+select release_lock('test');
+release_lock('test')
+1
+# RELEASE_LOCK returns NULL if it doesn't release a lock and there is no such lock
+select release_lock('test');
+release_lock('test')
+NULL
+# Test that get_lock() returns NULL if error.
+select get_lock('test', 0);
+get_lock('test', 0)
+1
+# -> Switching to connection 'con1'
+create table t1 select connection_id() as id;
+select get_lock('test', 7200);
+# -> Switching to connection 'default'
+select (@id := id) - id from t1;
+(@id := id) - id
+0
+kill query @id;
+# -> Switching to connection 'con1'
+get_lock('test', 7200)
+NULL
+# -> Switching to connection 'default'
+# GET_LOCK() works recursively 
+select get_lock('test', 0);
+get_lock('test', 0)
+1
+select get_lock('test', 0);
+get_lock('test', 0)
+1
+select get_lock('test', 0);
+get_lock('test', 0)
+1
+# RELEASE_LOCK() needs to be called recursively then, too
+select release_lock('test');
+release_lock('test')
+1
+select release_lock('test');
+release_lock('test')
+1
+select release_lock('test');
+release_lock('test')
+1
+# Once the last instance of the lock is released, 
+# the next call returns NULL
+select release_lock('test');
+release_lock('test')
+1
+# Multiple locks in the same session are OK
+select get_lock('test1', 0);
+get_lock('test1', 0)
+1
+select get_lock('test2', 0);
+get_lock('test2', 0)
+1
+select get_lock('test3', 0);
+get_lock('test3', 0)
+1
+select release_lock('test1');
+release_lock('test1')
+1
+select release_lock('test2');
+release_lock('test2')
+1
+select release_lock('test3');
+release_lock('test3')
+1
+# Deadlocks are detected e.g. in case of a mutual wait
+select get_lock('test1', 0);
+get_lock('test1', 0)
+1
+# -> Switching to connection 'con1'
+select get_lock('test2', 0);
+get_lock('test2', 0)
+1
+select get_lock('test1', 7200);
+# -> Switching to connection 'default'
+select get_lock('test2', 7200);
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+select release_lock('test1');
+release_lock('test1')
+1
+# -> Switching to connection 'con1'
+get_lock('test1', 7200)
+1
+select release_lock('test2');
+release_lock('test2')
+1
+select release_lock('test1');
+release_lock('test1')
+1
+# -> Switching to connection 'default'
+# LOCK/UNLOCK TABLES works fine with a user lock.
+lock table t1 write;
+select get_lock('test', 0);
+get_lock('test', 0)
+1
+unlock tables;
+commit;
+select release_lock('test');
+release_lock('test')
+1
+# GLOBAL READ LOCK works with fine with user locks
+select get_lock('test1', 0);
+get_lock('test1', 0)
+1
+flush tables with read lock;
+select get_lock('test2', 0);
+get_lock('test2', 0)
+1
+unlock tables;
+commit;
+select release_lock('test1');
+release_lock('test1')
+1
+select release_lock('test2');
+release_lock('test2')
+1
+# BEGIN/COMMIT/ROLLBACK don't unlock user locks.
+begin;
+select get_lock('test1', 0);
+get_lock('test1', 0)
+1
+select  get_lock('test2', 0);
+get_lock('test2', 0)
+1
+select count(*) from t1;
+count(*)
+1
+rollback;
+select release_lock('test1');
+release_lock('test1')
+1
+select release_lock('test2');
+release_lock('test2')
+1
+# Deadlocks between user locks and LOCK TABLES locks
+# are detected OK.
+select get_lock('test', 0);
+get_lock('test', 0)
+1
+# -> Switching to connection 'con1'
+lock table t1 write;
+select get_lock('test', 7200);
+# -> Switching to connection 'default'
+lock table t1 read;
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+select release_lock('test');
+release_lock('test')
+1
+# -> Switching to connection 'con1'
+get_lock('test', 7200)
+1
+select release_lock('test');
+release_lock('test')
+1
+unlock tables;
+# cleanup
+drop table t1;
+# check too long identifier names
+select get_lock(repeat('a', 192), 0);
+get_lock(repeat('a', 192), 0)
+1
+select is_used_lock(repeat('a', 192)) = connection_id();
+is_used_lock(repeat('a', 192)) = connection_id()
+1
+select is_free_lock(repeat('a', 192));
+is_free_lock(repeat('a', 192))
+0
+select release_lock(repeat('a', 192));
+release_lock(repeat('a', 192))
+1
+select get_lock(repeat('a', 193), 0);
+ERROR 42000: Identifier name 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' is too long
+select is_used_lock(repeat('a', 193));
+ERROR 42000: Identifier name 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' is too long
+select is_free_lock(repeat('a', 193));
+ERROR 42000: Identifier name 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' is too long
+select release_lock(repeat('a', 193));
+ERROR 42000: Identifier name 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' is too long
+#
 # End of 5.5 tests
 #
 #

mysql-test/suite/rpl/t/rpl_err_ignoredtable.test

@@ -43,7 +43,7 @@ insert into t3 values(connection_id());
 send update t2 set a = a + 1 + get_lock('crash_lock%20C', 10);
 
 connection master1;
-let $wait_condition= SELECT a > 1 FROM t2;
+let $wait_condition= SELECT count(*) > 0 FROM information_schema.processlist WHERE info LIKE 'update%' AND state='User lock';
 source include/wait_condition.inc;
 select (@id := id) - id from t3;
 kill @id;

mysql-test/suite/rpl/t/rpl_stm_000001.test

@@ -88,7 +88,8 @@ insert into t3 select get_lock('crash_lock%20C', 1) from t2;
 connection master;
 send update t1 set n = n + get_lock('crash_lock%20C', 2);
 connection master1;
-sleep 3;
+let $wait_condition= SELECT count(*) > 0 FROM information_schema.processlist WHERE info LIKE 'update%' AND state='User lock';
+source include/wait_condition.inc;
 select (@id := id) - id from t2;
 kill @id;
 # We don't drop t3 as this is a temporary table

mysql-test/t/func_misc.test

@@ -368,6 +368,183 @@ drop view  v_merge, vm;
 drop table t1,tv;
 
 
+--echo #
+--echo # GET_LOCK, RELEASE_LOCK, IS_USED_LOCK functions test
+--echo #
+
+--echo # IS_USED_LOCK, IS_FREE_LOCK: the  lock is not acquired
+--echo # Note: IS_USED_LOCK returns NULL if the lock is unused
+select is_used_lock('test');
+select is_free_lock('test');
+
+--echo # GET_LOCK returns 1 if it manages to acquire a lock
+select get_lock('test', 0);
+
+--echo # IS_USED_LOCK, IS_FREE_LOCK: the  lock is acquired
+select is_free_lock('test');
+select is_used_lock('test') = connection_id();
+
+connect (con1,localhost,root,,);
+--echo # -> Switching to connection 'con1'
+connection con1;
+--echo # IS_USED_LOCK, IS_FREE_LOCK: the  lock is acquired in another
+--echo # connection
+select is_used_lock('test') = connection_id();
+select is_free_lock('test');
+
+--echo # GET_LOCK returns 0 if it can't acquire a lock (wait timeout)
+select get_lock('test', 0);
+
+--echo # RELEASE_LOCK returns 0 if the lock belongs to another connection 
+select release_lock('test');
+
+--echo # -> Switching to connection 'default'
+connection default;
+
+--echo # RELEASE_LOCK returns 1 if it successfully releases a lock
+select release_lock('test');
+--echo # RELEASE_LOCK returns NULL if it doesn't release a lock and there is no such lock
+select release_lock('test');
+
+--echo # Test that get_lock() returns NULL if error.
+select get_lock('test', 0);
+--echo # -> Switching to connection 'con1'
+connection con1;
+create table t1 select connection_id() as id;
+send select get_lock('test', 7200);
+
+--echo # -> Switching to connection 'default'
+connection default;
+let $wait_condition= SELECT count(*) > 0 FROM information_schema.processlist WHERE info LIKE 'select%' AND state='User lock';
+source include/wait_condition.inc;
+select (@id := id) - id from t1;
+kill query @id;
+
+--echo # -> Switching to connection 'con1'
+connection con1;
+reap;
+
+--echo # -> Switching to connection 'default'
+connection default;
+
+--echo # GET_LOCK() works recursively 
+select get_lock('test', 0);
+select get_lock('test', 0);
+select get_lock('test', 0);
+
+--echo # RELEASE_LOCK() needs to be called recursively then, too
+select release_lock('test');
+select release_lock('test');
+select release_lock('test');
+
+--echo # Once the last instance of the lock is released, 
+--echo # the next call returns NULL
+select release_lock('test');
+
+
+--echo # Multiple locks in the same session are OK
+select get_lock('test1', 0);
+select get_lock('test2', 0);
+select get_lock('test3', 0);
+
+select release_lock('test1');
+select release_lock('test2');
+select release_lock('test3');
+
+--echo # Deadlocks are detected e.g. in case of a mutual wait
+select get_lock('test1', 0);
+
+--echo # -> Switching to connection 'con1'
+connection con1;
+select get_lock('test2', 0);
+send select get_lock('test1', 7200);
+
+--echo # -> Switching to connection 'default'
+connection default;
+let $wait_condition= SELECT count(*) > 0 FROM information_schema.processlist WHERE info LIKE 'select%' AND state='User lock';
+source include/wait_condition.inc;
+--error ER_LOCK_DEADLOCK
+select get_lock('test2', 7200);
+
+select release_lock('test1');
+
+--echo # -> Switching to connection 'con1'
+connection con1;
+reap;
+select release_lock('test2');
+select release_lock('test1');
+
+--echo # -> Switching to connection 'default'
+connection default;
+
+--echo # LOCK/UNLOCK TABLES works fine with a user lock.
+lock table t1 write;
+select get_lock('test', 0);
+unlock tables;
+commit;
+select release_lock('test');
+
+--echo # GLOBAL READ LOCK works with fine with user locks
+select get_lock('test1', 0);
+flush tables with read lock;
+select get_lock('test2', 0);
+unlock tables;
+commit;
+select release_lock('test1');
+select release_lock('test2');
+
+--echo # BEGIN/COMMIT/ROLLBACK don't unlock user locks.
+begin;
+select get_lock('test1', 0);
+select  get_lock('test2', 0);
+select count(*) from t1;
+rollback;
+select release_lock('test1');
+select release_lock('test2');
+
+--echo # Deadlocks between user locks and LOCK TABLES locks
+--echo # are detected OK.
+select get_lock('test', 0);
+
+--echo # -> Switching to connection 'con1'
+connection con1;
+lock table t1 write;
+send select get_lock('test', 7200);
+
+--echo # -> Switching to connection 'default'
+connection default;
+let $wait_condition= SELECT count(*) > 0 FROM information_schema.processlist WHERE info LIKE 'select%' AND state = 'User lock';
+source include/wait_condition.inc;
+--error ER_LOCK_DEADLOCK
+lock table t1 read;
+
+select release_lock('test');
+
+--echo # -> Switching to connection 'con1'
+connection con1;
+reap;
+select release_lock('test');
+unlock tables;
+
+--echo # cleanup
+disconnect con1;
+connection default;
+drop table t1;
+
+--echo # check too long identifier names
+select get_lock(repeat('a', 192), 0);
+select is_used_lock(repeat('a', 192)) = connection_id();
+select is_free_lock(repeat('a', 192));
+select release_lock(repeat('a', 192));
+--error ER_TOO_LONG_IDENT
+select get_lock(repeat('a', 193), 0);
+--error ER_TOO_LONG_IDENT
+select is_used_lock(repeat('a', 193));
+--error ER_TOO_LONG_IDENT
+select is_free_lock(repeat('a', 193));
+--error ER_TOO_LONG_IDENT
+select release_lock(repeat('a', 193));
+
 --echo #
 --echo # End of 5.5 tests
 --echo #

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();
 }
 

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
 {

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();
 
@@ -856,6 +859,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
@@ -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.
 */
 
@@ -2724,6 +2787,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.
 
@@ -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.

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;

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)

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,

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;
 }

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

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());
 
   /*