MDEV-5864 - Reduce usage of LOCK_open: TABLE_SHARE::tdc.free_tables

Let TABLE_SHARE::tdc.free_tables, TABLE_SHARE::tdc.all_tables,
TABLE_SHARE::tdc.flushed and corresponding invariants be protected by
per-share TABLE_SHARE::tdc.LOCK_table_share instead of global LOCK_open.

mysql-test/suite/perfschema/r/func_mutex.result

@@ -4,9 +4,9 @@ WHERE name LIKE 'wait/synch/mutex/%'
    OR name LIKE 'wait/synch/rwlock/%';
 flush status;
 select NAME from performance_schema.mutex_instances
-where NAME = 'wait/synch/mutex/sql/LOCK_open';
+where NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share' GROUP BY NAME;
 NAME
-wait/synch/mutex/sql/LOCK_open
+wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share
 select NAME from performance_schema.rwlock_instances
 where NAME = 'wait/synch/rwlock/sql/LOCK_grant';
 NAME
@@ -23,7 +23,7 @@ id	b
 1	initial value
 SET @before_count = (SELECT SUM(TIMER_WAIT)
 FROM performance_schema.events_waits_history_long
-WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 SELECT * FROM t1;
 id	b
 1	initial value
@@ -36,12 +36,12 @@ id	b
 8	initial value
 SET @after_count = (SELECT SUM(TIMER_WAIT)
 FROM performance_schema.events_waits_history_long
-WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 SELECT IF((@after_count - @before_count) > 0, 'Success', 'Failure') test_fm1_timed;
 test_fm1_timed
 Success
 UPDATE performance_schema.setup_instruments SET enabled = 'NO'
-WHERE NAME = 'wait/synch/mutex/sql/LOCK_open';
+WHERE NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share';
 TRUNCATE TABLE performance_schema.events_waits_history_long;
 TRUNCATE TABLE performance_schema.events_waits_history;
 TRUNCATE TABLE performance_schema.events_waits_current;
@@ -50,7 +50,7 @@ id	b
 1	initial value
 SET @before_count = (SELECT SUM(TIMER_WAIT)
 FROM performance_schema.events_waits_history_long
-WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 SELECT * FROM t1;
 id	b
 1	initial value
@@ -63,7 +63,7 @@ id	b
 8	initial value
 SET @after_count = (SELECT SUM(TIMER_WAIT)
 FROM performance_schema.events_waits_history_long
-WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 SELECT IF((COALESCE(@after_count, 0) - COALESCE(@before_count, 0)) = 0, 'Success', 'Failure') test_fm2_timed;
 test_fm2_timed
 Success

mysql-test/suite/perfschema/r/server_init.result

@@ -32,10 +32,6 @@ where name like "wait/synch/cond/mysys/THR_COND_threads";
 count(name)
 1
 select count(name) from mutex_instances
-where name like "wait/synch/mutex/sql/LOCK_open";
-count(name)
-1
-select count(name) from mutex_instances
 where name like "wait/synch/mutex/sql/LOCK_thread_count";
 count(name)
 1

mysql-test/suite/perfschema/t/func_mutex.test

@@ -18,7 +18,7 @@ flush status;
 
 # Make sure objects are instrumented
 select NAME from performance_schema.mutex_instances
-  where NAME = 'wait/synch/mutex/sql/LOCK_open';
+  where NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share' GROUP BY NAME;
 select NAME from performance_schema.rwlock_instances
   where NAME = 'wait/synch/rwlock/sql/LOCK_grant';
 
@@ -48,18 +48,18 @@ SELECT * FROM t1 WHERE id = 1;
 
 SET @before_count = (SELECT SUM(TIMER_WAIT)
                      FROM performance_schema.events_waits_history_long
-                     WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+                     WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 
 SELECT * FROM t1;
 
 SET @after_count = (SELECT SUM(TIMER_WAIT)
                     FROM performance_schema.events_waits_history_long
-                    WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+                    WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 
 SELECT IF((@after_count - @before_count) > 0, 'Success', 'Failure') test_fm1_timed;
 
 UPDATE performance_schema.setup_instruments SET enabled = 'NO'
-WHERE NAME = 'wait/synch/mutex/sql/LOCK_open';
+WHERE NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share';
 
 TRUNCATE TABLE performance_schema.events_waits_history_long;
 TRUNCATE TABLE performance_schema.events_waits_history;
@@ -69,13 +69,13 @@ SELECT * FROM t1 WHERE id = 1;
 
 SET @before_count = (SELECT SUM(TIMER_WAIT)
                      FROM performance_schema.events_waits_history_long
-                     WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+                     WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 
 SELECT * FROM t1;
 
 SET @after_count = (SELECT SUM(TIMER_WAIT)
                     FROM performance_schema.events_waits_history_long
-                    WHERE (EVENT_NAME = 'wait/synch/mutex/sql/LOCK_open'));
+                    WHERE (EVENT_NAME = 'wait/synch/mutex/sql/TABLE_SHARE::tdc.LOCK_table_share'));
 
 SELECT IF((COALESCE(@after_count, 0) - COALESCE(@before_count, 0)) = 0, 'Success', 'Failure') test_fm2_timed;
 

mysql-test/suite/perfschema/t/server_init.test

@@ -43,9 +43,6 @@ select count(name) from cond_instances
 
 # Verify that these global mutexes have been properly initilized in sql
 
-select count(name) from mutex_instances
- where name like "wait/synch/mutex/sql/LOCK_open";
-
 select count(name) from mutex_instances
  where name like "wait/synch/mutex/sql/LOCK_thread_count";
 

sql/lock.cc

@@ -790,7 +790,6 @@ MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count, uint flags)
   @param thd         Thread handle.
   @param db          The database name.
 
-  This function cannot be called while holding LOCK_open mutex.
   To avoid deadlocks, we do not try to obtain exclusive metadata
   locks in LOCK TABLES mode, since in this mode there may be
   other metadata locks already taken by the current connection,
@@ -842,9 +841,7 @@ bool lock_schema_name(THD *thd, const char *db)
   @param name        Object name in the schema.
 
   This function assumes that no metadata locks were acquired
-  before calling it. Additionally, it cannot be called while
-  holding LOCK_open mutex. Both these invariants are enforced by
-  asserts in MDL_context::acquire_locks().
+  before calling it. It is enforced by asserts in MDL_context::acquire_locks().
   To avoid deadlocks, we do not try to obtain exclusive metadata
   locks in LOCK TABLES mode, since in this mode there may be
   other metadata locks already taken by the current connection,

sql/mdl.cc

@@ -1903,8 +1903,6 @@ bool MDL_lock::has_pending_conflicting_lock(enum_mdl_type type)
 {
   bool result;
 
-  mysql_mutex_assert_not_owner(&LOCK_open);
-
   mysql_prlock_rdlock(&m_rwlock);
   result= (m_waiting.bitmap() & incompatible_granted_types_bitmap()[type]);
   mysql_prlock_unlock(&m_rwlock);
@@ -2079,7 +2077,6 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
 
   /* Don't take chances in production. */
   mdl_request->ticket= NULL;
-  mysql_mutex_assert_not_owner(&LOCK_open);
 
   /*
     Check whether the context already holds a shared lock on the object,
@@ -2169,7 +2166,6 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
 {
   MDL_ticket *ticket;
 
-  mysql_mutex_assert_not_owner(&LOCK_open);
   /*
     By submitting mdl_request->type to MDL_ticket::create()
     we effectively downgrade the cloned lock to the level of
@@ -2830,7 +2826,6 @@ void MDL_context::release_lock(enum_mdl_duration duration, MDL_ticket *ticket)
                        lock->key.db_name(), lock->key.name()));
 
   DBUG_ASSERT(this == ticket->get_ctx());
-  mysql_mutex_assert_not_owner(&LOCK_open);
 
   lock->remove_ticket(&MDL_lock::m_granted, ticket);
 
@@ -2923,8 +2918,6 @@ void MDL_context::release_all_locks_for_name(MDL_ticket *name)
 
 void MDL_ticket::downgrade_lock(enum_mdl_type type)
 {
-  mysql_mutex_assert_not_owner(&LOCK_open);
-
   /*
     Do nothing if already downgraded. Used when we FLUSH TABLE under
     LOCK TABLES and a table is listed twice in LOCK TABLES list.

sql/mdl.h

@@ -519,17 +519,7 @@ public:
 
   virtual bool inspect_edge(MDL_context *dest) = 0;
   virtual ~MDL_wait_for_graph_visitor();
-  MDL_wait_for_graph_visitor() :m_lock_open_count(0) {}
-public:
-  /**
-   XXX, hack: During deadlock search, we may need to
-   inspect TABLE_SHAREs and acquire LOCK_open. Since
-   LOCK_open is not a recursive mutex, count here how many
-   times we "took" it (but only take and release once).
-   Not using a native recursive mutex or rwlock in 5.5 for
-   LOCK_open since it has significant performance impacts.
-  */
-  uint m_lock_open_count;
+  MDL_wait_for_graph_visitor() {}
 };
 
 /**
@@ -980,10 +970,6 @@ extern "C" unsigned long thd_get_thread_id(const MYSQL_THD thd);
 */
 extern "C" int thd_is_connected(MYSQL_THD thd);
 
-#ifndef DBUG_OFF
-extern mysql_mutex_t LOCK_open;
-#endif
-
 
 /*
   Start-up parameter for the maximum size of the unused MDL_lock objects cache

sql/sql_base.cc

@@ -262,7 +262,7 @@ uint get_table_def_key(const TABLE_LIST *table_list, const char **key)
   NOTES
     One gets only a list of tables for which one has any kind of privilege.
     db and table names are allocated in result struct, so one doesn't need
-    a lock on LOCK_open when traversing the return list.
+    a lock when traversing the return list.
 
   RETURN VALUES
     NULL	Error (Probably OOM)
@@ -308,13 +308,13 @@ OPEN_TABLE_LIST *list_open_tables(THD *thd, const char *db, const char *wild)
 		  share->db.str)+1,
 	   share->table_name.str);
     (*start_list)->in_use= 0;
-    mysql_mutex_lock(&LOCK_open);
+    mysql_mutex_lock(&share->tdc.LOCK_table_share);
     TABLE_SHARE::All_share_tables_list::Iterator it(share->tdc.all_tables);
     TABLE *table;
     while ((table= it++))
       if (table->in_use)
         ++(*start_list)->in_use;
-    mysql_mutex_unlock(&LOCK_open);
+    mysql_mutex_unlock(&share->tdc.LOCK_table_share);
     (*start_list)->locked= 0;                   /* Obsolete. */
     start_list= &(*start_list)->next;
     *start_list=0;
@@ -335,7 +335,6 @@ void intern_close_table(TABLE *table)
                         table->s ? table->s->db.str : "?",
                         table->s ? table->s->table_name.str : "?",
                         (long) table));
-  mysql_mutex_assert_not_owner(&LOCK_open);
 
   free_io_cache(table);
   delete table->triggers;
@@ -368,7 +367,7 @@ void free_io_cache(TABLE *table)
 
    @param share Table share.
 
-   @pre Caller should have LOCK_open mutex.
+   @pre Caller should have TABLE_SHARE::tdc.LOCK_table_share mutex.
 */
 
 void kill_delayed_threads_for_table(TABLE_SHARE *share)
@@ -376,7 +375,7 @@ void kill_delayed_threads_for_table(TABLE_SHARE *share)
   TABLE_SHARE::All_share_tables_list::Iterator it(share->tdc.all_tables);
   TABLE *tab;
 
-  mysql_mutex_assert_owner(&LOCK_open);
+  mysql_mutex_assert_owner(&share->tdc.LOCK_table_share);
 
   if (!delayed_insert_threads)
     return;
@@ -774,8 +773,6 @@ static void mark_used_tables_as_free_for_reuse(THD *thd, TABLE *table)
 
 static void close_open_tables(THD *thd)
 {
-  mysql_mutex_assert_not_owner(&LOCK_open);
-
   DBUG_PRINT("info", ("thd->open_tables: 0x%lx", (long) thd->open_tables));
 
   while (thd->open_tables)
@@ -822,7 +819,6 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
 
   memcpy(key, share->table_cache_key.str, key_length);
 
-  mysql_mutex_assert_not_owner(&LOCK_open);
   for (TABLE **prev= &thd->open_tables; *prev; )
   {
     TABLE *table= *prev;
@@ -1018,7 +1014,6 @@ void close_thread_table(THD *thd, TABLE **table_ptr)
                         table->s->table_name.str, (long) table));
   DBUG_ASSERT(table->key_read == 0);
   DBUG_ASSERT(!table->file || table->file->inited == handler::NONE);
-  mysql_mutex_assert_not_owner(&LOCK_open);
 
   /*
     The metadata lock must be released after giving back
@@ -1049,7 +1044,10 @@ void close_thread_table(THD *thd, TABLE **table_ptr)
     table->file->ha_reset();
   }
 
-  /* Do this *before* entering the LOCK_open critical section. */
+  /*
+    Do this *before* entering the TABLE_SHARE::tdc.LOCK_table_share
+    critical section.
+  */
   if (table->file != NULL)
     table->file->unbind_psi();
 

sql/sql_handler.cc

@@ -1115,8 +1115,6 @@ void mysql_ha_flush(THD *thd)
   SQL_HANDLER *hash_tables;
   DBUG_ENTER("mysql_ha_flush");
 
-  mysql_mutex_assert_not_owner(&LOCK_open);
-
   /*
     Don't try to flush open HANDLERs when we're working with
     system tables. The main MDL context is backed up and we can't

sql/sql_test.cc

@@ -88,9 +88,9 @@ static void print_cached_tables(void)
   puts("DB             Table                            Version  Thread  Open  Lock");
 
   tdc_it.init();
-  mysql_mutex_lock(&LOCK_open);
   while ((share= tdc_it.next()))
   {
+    mysql_mutex_lock(&share->tdc.LOCK_table_share);
     TABLE_SHARE::All_share_tables_list::Iterator it(share->tdc.all_tables);
     while ((entry= it++))
     {
@@ -102,8 +102,8 @@ static void print_cached_tables(void)
              in_use ? lock_descriptions[(int)entry->reginfo.lock_type] :
                       "Not in use");
     }
+    mysql_mutex_unlock(&share->tdc.LOCK_table_share);
   }
-  mysql_mutex_unlock(&LOCK_open);
   tdc_it.deinit();
   printf("\nCurrent refresh version: %ld\n", tdc_refresh_version());
   fflush(stdout);

sql/table.cc

@@ -3805,13 +3805,14 @@ bool TABLE_SHARE::visit_subgraph(Wait_for_flush *wait_for_flush,
 
   /*
     To protect all_tables list from being concurrently modified
-    while we are iterating through it we acquire LOCK_open.
+    while we are iterating through it we increment tdc.all_tables_refs.
     This does not introduce deadlocks in the deadlock detector
-    because we won't try to acquire LOCK_open while
+    because we won't try to acquire tdc.LOCK_table_share while
     holding a write-lock on MDL_lock::m_rwlock.
   */
-  if (gvisitor->m_lock_open_count++ == 0)
-    mysql_mutex_lock(&LOCK_open);
+  mysql_mutex_lock(&tdc.LOCK_table_share);
+  tdc.all_tables_refs++;
+  mysql_mutex_unlock(&tdc.LOCK_table_share);
 
   All_share_tables_list::Iterator tables_it(tdc.all_tables);
 
@@ -3854,8 +3855,10 @@ end_leave_node:
   gvisitor->leave_node(src_ctx);
 
 end:
-  if (gvisitor->m_lock_open_count-- == 1)
-    mysql_mutex_unlock(&LOCK_open);
+  mysql_mutex_lock(&tdc.LOCK_table_share);
+  if (!--tdc.all_tables_refs)
+    mysql_cond_broadcast(&tdc.COND_release);
+  mysql_mutex_unlock(&tdc.LOCK_table_share);
 
   return result;
 }

sql/table.h

@@ -616,12 +616,14 @@ struct TABLE_SHARE
   struct
   {
     /**
-      Protects ref_count and m_flush_tickets.
+      Protects ref_count, m_flush_tickets, all_tables, free_tables, flushed,
+      all_tables_refs.
     */
     mysql_mutex_t LOCK_table_share;
     mysql_cond_t COND_release;
     TABLE_SHARE *next, **prev;            /* Link to unused shares */
     uint ref_count;                       /* How many TABLE objects uses this */
+    uint all_tables_refs;                 /* Number of refs to all_tables */
     /**
       List of tickets representing threads waiting for the share to be flushed.
     */

sql/table_cache.cc

@@ -69,13 +69,6 @@ static bool tdc_inited;
 static int32 tc_count; /**< Number of TABLE objects in table cache. */
 
 
-/**
-  Protects TABLE_SHARE::tdc.free_tables, TABLE_SHARE::tdc.all_tables.
-*/
-
-mysql_mutex_t LOCK_open;
-
-
 /**
   Protects unused shares list.
 
@@ -90,11 +83,9 @@ static mysql_rwlock_t LOCK_tdc; /**< Protects tdc_hash. */
 my_atomic_rwlock_t LOCK_tdc_atomics; /**< Protects tdc_version. */
 
 #ifdef HAVE_PSI_INTERFACE
-static PSI_mutex_key key_LOCK_open, key_LOCK_unused_shares,
-                     key_TABLE_SHARE_LOCK_table_share;
+static PSI_mutex_key key_LOCK_unused_shares, key_TABLE_SHARE_LOCK_table_share;
 static PSI_mutex_info all_tc_mutexes[]=
 {
-  { &key_LOCK_open, "LOCK_open", PSI_FLAG_GLOBAL },
   { &key_LOCK_unused_shares, "LOCK_unused_shares", PSI_FLAG_GLOBAL },
   { &key_TABLE_SHARE_LOCK_table_share, "TABLE_SHARE::tdc.LOCK_table_share", 0 }
 };
@@ -170,6 +161,33 @@ static void tc_remove_table(TABLE *table)
 }
 
 
+/**
+  Wait for MDL deadlock detector to complete traversing tdc.all_tables.
+
+  Must be called before updating TABLE_SHARE::tdc.all_tables.
+*/
+
+static void tc_wait_for_mdl_deadlock_detector(TABLE_SHARE *share)
+{
+  while (share->tdc.all_tables_refs)
+    mysql_cond_wait(&share->tdc.COND_release, &share->tdc.LOCK_table_share);
+}
+
+
+/**
+  Get last element of tdc.free_tables.
+*/
+
+static TABLE *tc_free_tables_back(TABLE_SHARE *share)
+{
+  TABLE_SHARE::TABLE_list::Iterator it(share->tdc.free_tables);
+  TABLE *entry, *last= 0;
+   while ((entry= it++))
+     last= entry;
+  return last;
+}
+
+
 /**
   Free all unused TABLE objects.
 
@@ -193,9 +211,11 @@ void tc_purge(bool mark_flushed)
   TABLE_SHARE::TABLE_list purge_tables;
 
   tdc_it.init();
-  mysql_mutex_lock(&LOCK_open);
   while ((share= tdc_it.next()))
   {
+    mysql_mutex_lock(&share->tdc.LOCK_table_share);
+    tc_wait_for_mdl_deadlock_detector(share);
+
     if (mark_flushed)
       share->tdc.flushed= true;
     while ((table= share->tdc.free_tables.pop_front()))
@@ -203,9 +223,9 @@ void tc_purge(bool mark_flushed)
       tc_remove_table(table);
       purge_tables.push_front(table);
     }
+    mysql_mutex_unlock(&share->tdc.LOCK_table_share);
   }
   tdc_it.deinit();
-  mysql_mutex_unlock(&LOCK_open);
 
   while ((table= purge_tables.pop_front()))
     intern_close_table(table);
@@ -232,9 +252,10 @@ void tc_add_table(THD *thd, TABLE *table)
 {
   bool need_purge;
   DBUG_ASSERT(table->in_use == thd);
-  mysql_mutex_lock(&LOCK_open);
+  mysql_mutex_lock(&table->s->tdc.LOCK_table_share);
+  tc_wait_for_mdl_deadlock_detector(table->s);
   table->s->tdc.all_tables.push_front(table);
-  mysql_mutex_unlock(&LOCK_open);
+  mysql_mutex_unlock(&table->s->tdc.LOCK_table_share);
 
   /* If we have too many TABLE instances around, try to get rid of them */
   my_atomic_rwlock_wrlock(&LOCK_tdc_atomics);
@@ -243,31 +264,48 @@ void tc_add_table(THD *thd, TABLE *table)
 
   if (need_purge)
   {
-    TABLE *purge_table= 0;
+    TABLE_SHARE *purge_share= 0;
     TABLE_SHARE *share;
+    TABLE *entry;
+    ulonglong purge_time;
     TDC_iterator tdc_it;
 
     tdc_it.init();
-    mysql_mutex_lock(&LOCK_open);
     while ((share= tdc_it.next()))
     {
-      TABLE_SHARE::TABLE_list::Iterator it(share->tdc.free_tables);
-      TABLE *entry;
-      while ((entry= it++))
-        if (!purge_table || entry->tc_time < purge_table->tc_time)
-          purge_table= entry;
+      mysql_mutex_lock(&share->tdc.LOCK_table_share);
+      if ((entry= tc_free_tables_back(share)) &&
+          (!purge_share || entry->tc_time < purge_time))
+      {
+          purge_share= share;
+          purge_time= entry->tc_time;
+      }
+      mysql_mutex_unlock(&share->tdc.LOCK_table_share);
     }
-    tdc_it.deinit();
 
-    if (purge_table)
+    if (purge_share)
     {
-      purge_table->s->tdc.free_tables.remove(purge_table);
-      tc_remove_table(purge_table);
-      mysql_mutex_unlock(&LOCK_open);
-      intern_close_table(purge_table);
+      mysql_mutex_lock(&purge_share->tdc.LOCK_table_share);
+      tc_wait_for_mdl_deadlock_detector(purge_share);
+      tdc_it.deinit();
+      /*
+        It may happen that oldest table was acquired meanwhile. In this case
+        just go ahead, number of objects in table cache will normalize
+        eventually.
+      */
+      if ((entry= tc_free_tables_back(purge_share)) &&
+          entry->tc_time == purge_time)
+      {
+        entry->s->tdc.free_tables.remove(entry);
+        tc_remove_table(entry);
+        mysql_mutex_unlock(&purge_share->tdc.LOCK_table_share);
+        intern_close_table(entry);
+      }
+      else
+        mysql_mutex_unlock(&purge_share->tdc.LOCK_table_share);
     }
     else
-      mysql_mutex_unlock(&LOCK_open);
+      tdc_it.deinit();
   }
 }
 
@@ -292,9 +330,9 @@ static TABLE *tc_acquire_table(THD *thd, TABLE_SHARE *share)
 {
   TABLE *table;
 
-  mysql_mutex_lock(&LOCK_open);
+  mysql_mutex_lock(&share->tdc.LOCK_table_share);
   table= share->tdc.free_tables.pop_front();
-  mysql_mutex_unlock(&LOCK_open);
+  mysql_mutex_unlock(&share->tdc.LOCK_table_share);
 
   if (table)
   {
@@ -328,7 +366,7 @@ static TABLE *tc_acquire_table(THD *thd, TABLE_SHARE *share)
   @note Another thread may mark share for purge any moment (even
   after version check). It means to-be-purged object may go to
   unused lists. This other thread is expected to call tc_purge(),
-  which is synchronized with us on LOCK_open.
+  which is synchronized with us on TABLE_SHARE::tdc.LOCK_table_share.
 
   @return
     @retval true  object purged
@@ -342,17 +380,17 @@ bool tc_release_table(TABLE *table)
 
   if (table->needs_reopen() || tc_records() > tc_size)
   {
-    mysql_mutex_lock(&LOCK_open);
+    mysql_mutex_lock(&table->s->tdc.LOCK_table_share);
     goto purge;
   }
 
   table->tc_time= my_interval_timer();
 
-  mysql_mutex_lock(&LOCK_open);
+  mysql_mutex_lock(&table->s->tdc.LOCK_table_share);
   if (table->s->tdc.flushed)
     goto purge;
   /*
-    in_use doesn't really need protection of LOCK_open, but must be reset after
+    in_use doesn't really need mutex protection, but must be reset after
     checking tdc.flushed and before this table appears in free_tables.
     Resetting in_use is needed only for print_cached_tables() and
     list_open_tables().
@@ -360,12 +398,13 @@ bool tc_release_table(TABLE *table)
   table->in_use= 0;
   /* Add table to the list of unused TABLE objects for this share. */
   table->s->tdc.free_tables.push_front(table);
-  mysql_mutex_unlock(&LOCK_open);
+  mysql_mutex_unlock(&table->s->tdc.LOCK_table_share);
   return false;
 
 purge:
+  tc_wait_for_mdl_deadlock_detector(table->s);
   tc_remove_table(table);
-  mysql_mutex_unlock(&LOCK_open);
+  mysql_mutex_unlock(&table->s->tdc.LOCK_table_share);
   table->in_use= 0;
   intern_close_table(table);
   return true;
@@ -446,13 +485,6 @@ int tdc_init(void)
   init_tc_psi_keys();
 #endif
   tdc_inited= true;
-  mysql_mutex_init(key_LOCK_open, &LOCK_open, MY_MUTEX_INIT_FAST);
-  mysql_mutex_record_order(&LOCK_active_mi, &LOCK_open);
-  /*
-    We must have LOCK_open before LOCK_global_system_variables because
-    LOCK_open is held while sql_plugin.cc::intern_sys_var_ptr() is called.
-  */
-  mysql_mutex_record_order(&LOCK_open, &LOCK_global_system_variables);
   mysql_mutex_init(key_LOCK_unused_shares, &LOCK_unused_shares,
                    MY_MUTEX_INIT_FAST);
   mysql_rwlock_init(key_rwlock_LOCK_tdc, &LOCK_tdc);
@@ -505,7 +537,6 @@ void tdc_deinit(void)
     my_atomic_rwlock_destroy(&LOCK_tdc_atomics);
     mysql_rwlock_destroy(&LOCK_tdc);
     mysql_mutex_destroy(&LOCK_unused_shares);
-    mysql_mutex_destroy(&LOCK_open);
   }
   DBUG_VOID_RETURN;
 }
@@ -575,6 +606,7 @@ void tdc_init_share(TABLE_SHARE *share)
   tdc_assign_new_table_id(share);
   share->tdc.version= tdc_refresh_version();
   share->tdc.flushed= false;
+  share->tdc.all_tables_refs= 0;
   DBUG_VOID_RETURN;
 }
 
@@ -590,6 +622,7 @@ void tdc_deinit_share(TABLE_SHARE *share)
   DBUG_ASSERT(share->tdc.m_flush_tickets.is_empty());
   DBUG_ASSERT(share->tdc.all_tables.is_empty());
   DBUG_ASSERT(share->tdc.free_tables.is_empty());
+  DBUG_ASSERT(share->tdc.all_tables_refs == 0);
   mysql_cond_destroy(&share->tdc.COND_release);
   mysql_mutex_destroy(&share->tdc.LOCK_table_share);
   DBUG_VOID_RETURN;
@@ -826,7 +859,8 @@ void tdc_release_share(TABLE_SHARE *share)
   if (share->tdc.ref_count > 1)
   {
     share->tdc.ref_count--;
-    mysql_cond_broadcast(&share->tdc.COND_release);
+    if (!share->is_view)
+      mysql_cond_broadcast(&share->tdc.COND_release);
     mysql_mutex_unlock(&share->tdc.LOCK_table_share);
     DBUG_VOID_RETURN;
   }
@@ -954,13 +988,14 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
     I_P_List <TABLE, TABLE_share> purge_tables;
     uint my_refs= 1;
 
-    mysql_mutex_lock(&LOCK_open);
+    mysql_mutex_lock(&share->tdc.LOCK_table_share);
+    tc_wait_for_mdl_deadlock_detector(share);
     /*
-      Set share's version to zero in order to ensure that it gets
+      Mark share flushed in order to ensure that it gets
       automatically deleted once it is no longer referenced.
 
       Note that code in TABLE_SHARE::wait_for_old_version() assumes that
-      incrementing of refresh_version is followed by purge of unused table
+      marking share flushed is followed by purge of unused table
       shares.
     */
     if (remove_type != TDC_RT_REMOVE_NOT_OWN_KEEP_SHARE)
@@ -985,7 +1020,7 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
       }
     }
     DBUG_ASSERT(share->tdc.all_tables.is_empty() || remove_type != TDC_RT_REMOVE_ALL);
-    mysql_mutex_unlock(&LOCK_open);
+    mysql_mutex_unlock(&share->tdc.LOCK_table_share);
 
     while ((table= purge_tables.pop_front()))
       intern_close_table(table);

sql/table_cache.h

@@ -26,7 +26,6 @@ enum enum_tdc_remove_table_type
 
 extern ulong tdc_size;
 extern ulong tc_size;
-extern mysql_mutex_t LOCK_open; /* FIXME: make private */
 
 extern int tdc_init(void);
 extern void tdc_start_shutdown(void);

storage/federated/ha_federated.cc

@@ -3119,14 +3119,6 @@ int ha_federated::real_connect()
   String sql_query(buffer, sizeof(buffer), &my_charset_bin);
   DBUG_ENTER("ha_federated::real_connect");
 
-  /* 
-    Bug#25679
-    Ensure that we do not hold the LOCK_open mutex while attempting
-    to establish Federated connection to guard against a trivial
-    Denial of Service scenerio.
-  */
-  mysql_mutex_assert_not_owner(&LOCK_open);
-
   DBUG_ASSERT(mysql == NULL);
 
   if (!(mysql= mysql_init(NULL)))

storage/federatedx/ha_federatedx.cc

@@ -3384,14 +3384,6 @@ int ha_federatedx::create(const char *name, TABLE *table_arg,
   {
     FEDERATEDX_SERVER server;
 
-    /* 
-      Bug#25679
-      Ensure that we do not hold the LOCK_open mutex while attempting
-      to establish FederatedX connection to guard against a trivial
-      Denial of Service scenerio.
-    */
-    mysql_mutex_assert_not_owner(&LOCK_open);
-
     fill_server(thd->mem_root, &server, &tmp_share, create_info->table_charset);
 
 #ifndef DBUG_OFF