MDEV-5492 - Reduce usage of LOCK_open: TABLE::in_use

Move TABLE::in_use out of LOCK_open.

This is done with assumtion that foreign threads accessing TABLE::in_use
will only need consistent value _after_ marking table for flush and purging
unused table instances. In this case TABLE::in_use will always point to a
valid thread object.

Previously FLUSH TABLES thread may wait for tables flushed subsequently by
concurrent threads which breaks the above assumption, e.g.:
open tables: t1 (version= 1)
thr1 (FLUSH TABLES): refresh_version++
thr1 (FLUSH TABLES): purge table cache
open tables: none
thr2 (SELECT * FROM t1): open tables: t1
open tables: t1 (version= 2)
thr2 (FLUSH TABLES): refresh_version++
thr2 (FLUSH TABLES): purge table cache
thr1 (FLUSH TABLES): wait for old tables (including t1 with version 2)

It is fixed so that FLUSH TABLES waits only for tables that were open
heretofore.

sql/sql_base.cc

@@ -385,7 +385,8 @@ void kill_delayed_threads_for_table(TABLE_SHARE *share)
   {
     THD *in_use= tab->in_use;
 
-    if (in_use && (in_use->system_thread & SYSTEM_THREAD_DELAYED_INSERT) &&
+    DBUG_ASSERT(in_use && tab->s->tdc.flushed);
+    if ((in_use->system_thread & SYSTEM_THREAD_DELAYED_INSERT) &&
         ! in_use->killed)
     {
       in_use->killed= KILL_SYSTEM_THREAD;
@@ -426,9 +427,12 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
 {
   bool result= FALSE;
   struct timespec abstime;
+  ulong refresh_version;
   DBUG_ENTER("close_cached_tables");
   DBUG_ASSERT(thd || (!wait_for_refresh && !tables));
 
+  refresh_version= tdc_increment_refresh_version();
+
   if (!tables)
   {
     /*
@@ -438,13 +442,12 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
       incrementing of refresh_version is followed by purge of unused table
       shares.
     */
-    tdc_increment_refresh_version();
     kill_delayed_threads();
     /*
       Get rid of all unused TABLE and TABLE_SHARE instances. By doing
       this we automatically close all tables which were marked as "old".
     */
-    tc_purge();
+    tc_purge(true);
     /* Free table shares which were not freed implicitly by loop above. */
     tdc_purge(true);
   }
@@ -526,7 +529,7 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
       while ((share= tdc_it.next()))
       {
         mysql_mutex_lock(&share->tdc.LOCK_table_share);
-        if (share->has_old_version())
+        if (share->tdc.flushed && share->tdc.version < refresh_version)
         {
           /* wait_for_old_version() will unlock mutex and free share */
           found= true;
@@ -554,7 +557,8 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
       if (thd->killed)
         break;
       if (tdc_wait_for_old_version(thd, table->db, table->table_name, timeout,
-                                   MDL_wait_for_subgraph::DEADLOCK_WEIGHT_DDL))
+                                   MDL_wait_for_subgraph::DEADLOCK_WEIGHT_DDL,
+                                   refresh_version))
       {
         result= TRUE;
         break;
@@ -1754,7 +1758,7 @@ bool wait_while_table_is_used(THD *thd, TABLE *table,
   DBUG_ENTER("wait_while_table_is_used");
   DBUG_PRINT("enter", ("table: '%s'  share: 0x%lx  db_stat: %u  version: %lu",
                        table->s->table_name.str, (ulong) table->s,
-                       table->db_stat, table->s->version));
+                       table->db_stat, table->s->tdc.version));
 
   if (thd->mdl_context.upgrade_shared_lock(
              table->mdl_ticket, MDL_EXCLUSIVE,
@@ -2321,7 +2325,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
 
   /*
     Check if this TABLE_SHARE-object corresponds to a view. Note, that there is
-    no need to call TABLE_SHARE::has_old_version() as we do for regular tables,
+    no need to check TABLE_SHARE::tdc.flushed as we do for regular tables,
     because view shares are always up to date.
   */
   if (share->is_view)
@@ -2362,7 +2366,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
 
   if (!(flags & MYSQL_OPEN_IGNORE_FLUSH))
   {
-    if (share->has_old_version())
+    if (share->tdc.flushed)
     {
       /*
         We already have an MDL lock. But we have encountered an old
@@ -2394,7 +2398,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
       goto retry_share;
     }
 
-    if (thd->open_tables && thd->open_tables->s->version != share->version)
+    if (thd->open_tables && thd->open_tables->s->tdc.flushed)
     {
       /*
         If the version changes while we're opening the tables,

sql/sql_handler.cc

@@ -1135,7 +1135,7 @@ void mysql_ha_flush(THD *thd)
         ((hash_tables->table->mdl_ticket &&
          hash_tables->table->mdl_ticket->has_pending_conflicting_lock()) ||
          (!hash_tables->table->s->tmp_table &&
-          hash_tables->table->s->has_old_version())))
+          hash_tables->table->s->tdc.flushed)))
       mysql_ha_close_table(hash_tables);
   }
 

sql/sql_insert.cc

@@ -3072,7 +3072,7 @@ bool Delayed_insert::handle_inserts(void)
 
   THD_STAGE_INFO(&thd, stage_insert);
   max_rows= delayed_insert_limit;
-  if (thd.killed || table->s->has_old_version())
+  if (thd.killed || table->s->tdc.flushed)
   {
     thd.killed= KILL_SYSTEM_THREAD;
     max_rows= ULONG_MAX;                     // Do as much as possible

sql/sql_test.cc

@@ -94,11 +94,12 @@ static void print_cached_tables(void)
     TABLE_SHARE::All_share_tables_list::Iterator it(share->tdc.all_tables);
     while ((entry= it++))
     {
+      THD *in_use= entry->in_use;
       printf("%-14.14s %-32s%6ld%8ld%6d  %s\n",
-             entry->s->db.str, entry->s->table_name.str, entry->s->version,
-             entry->in_use ? entry->in_use->thread_id : 0,
+             entry->s->db.str, entry->s->table_name.str, entry->s->tdc.version,
+             in_use ? in_use->thread_id : 0,
              entry->db_stat ? 1 : 0,
-             entry->in_use ? lock_descriptions[(int)entry->reginfo.lock_type] :
+             in_use ? lock_descriptions[(int)entry->reginfo.lock_type] :
                       "Not in use");
     }
   }

sql/table.cc

@@ -3805,7 +3805,7 @@ bool TABLE_SHARE::visit_subgraph(Wait_for_flush *wait_for_flush,
   bool result= TRUE;
 
   /*
-    To protect used_tables list from being concurrently modified
+    To protect all_tables list from being concurrently modified
     while we are iterating through it we acquire LOCK_open.
     This does not introduce deadlocks in the deadlock detector
     because we won't try to acquire LOCK_open while
@@ -3832,7 +3832,8 @@ bool TABLE_SHARE::visit_subgraph(Wait_for_flush *wait_for_flush,
 
   while ((table= tables_it++))
   {
-    if (table->in_use && gvisitor->inspect_edge(&table->in_use->mdl_context))
+    DBUG_ASSERT(table->in_use && tdc.flushed);
+    if (gvisitor->inspect_edge(&table->in_use->mdl_context))
     {
       goto end_leave_node;
     }
@@ -3841,7 +3842,8 @@ bool TABLE_SHARE::visit_subgraph(Wait_for_flush *wait_for_flush,
   tables_it.rewind();
   while ((table= tables_it++))
   {
-    if (table->in_use && table->in_use->mdl_context.visit_subgraph(gvisitor))
+    DBUG_ASSERT(table->in_use && tdc.flushed);
+    if (table->in_use->mdl_context.visit_subgraph(gvisitor))
     {
       goto end_leave_node;
     }
@@ -3890,7 +3892,7 @@ bool TABLE_SHARE::wait_for_old_version(THD *thd, struct timespec *abstime,
   MDL_wait::enum_wait_status wait_status;
 
   mysql_mutex_assert_owner(&tdc.LOCK_table_share);
-  DBUG_ASSERT(has_old_version());
+  DBUG_ASSERT(tdc.flushed);
 
   tdc.m_flush_tickets.push_front(&ticket);
 

sql/table.h

@@ -481,8 +481,6 @@ TABLE_CATEGORY get_table_category(const LEX_STRING *db,
 struct TABLE_share;
 struct All_share_tables;
 
-extern ulong tdc_refresh_version(void);
-
 typedef struct st_table_field_type
 {
   LEX_STRING name;
@@ -623,6 +621,8 @@ struct TABLE_SHARE
     */
     All_share_tables_list all_tables;
     TABLE_list free_tables;
+    ulong version;
+    bool flushed;
   } tdc;
 
   LEX_CUSTRING tabledef_version;
@@ -668,7 +668,6 @@ struct TABLE_SHARE
   key_map keys_for_keyread;
   ha_rows min_rows, max_rows;		/* create information */
   ulong   avg_row_length;		/* create information */
-  ulong   version;
   ulong   mysql_version;		/* 0 if .frm is created before 5.0 */
   ulong   reclength;			/* Recordlength */
   /* Stored record length. No generated-only virtual fields are included */
@@ -847,12 +846,6 @@ struct TABLE_SHARE
     return table_map_id;
   }
 
-  /** Is this table share being expelled from the table definition cache?  */
-  inline bool has_old_version() const
-  {
-    return version != tdc_refresh_version();
-  }
-
   /**
     Convert unrelated members of TABLE_SHARE to one enum
     representing its type.

sql/table_cache.cc

@@ -44,6 +44,8 @@
 
   Table cache invariants:
   - TABLE_SHARE::free_tables shall not contain objects with TABLE::in_use != 0
+  - TABLE_SHARE::free_tables shall not receive new objects if
+    TABLE_SHARE::tdc.flushed is true
 */
 
 #include "my_global.h"
@@ -68,8 +70,7 @@ static int32 tc_count; /**< Number of TABLE objects in table cache. */
 
 
 /**
-  Protects TABLE_SHARE::tdc.free_tables, TABLE_SHARE::tdc.all_tables,
-  TABLE::in_use.
+  Protects TABLE_SHARE::tdc.free_tables, TABLE_SHARE::tdc.all_tables.
 */
 
 mysql_mutex_t LOCK_open;
@@ -176,7 +177,7 @@ static void tc_remove_table(TABLE *table)
         periodicly flush all not used tables.
 */
 
-void tc_purge(void)
+void tc_purge(bool mark_flushed)
 {
   TABLE_SHARE *share;
   TABLE *table;
@@ -187,6 +188,8 @@ void tc_purge(void)
   mysql_mutex_lock(&LOCK_open);
   while ((share= tdc_it.next()))
   {
+    if (mark_flushed)
+      share->tdc.flushed= true;
     while ((table= share->tdc.free_tables.pop_front()))
     {
       tc_remove_table(table);
@@ -203,49 +206,6 @@ void tc_purge(void)
 }
 
 
-/**
-  Verify consistency of used/unused lists (for debugging).
-*/
-
-#ifdef EXTRA_DEBUG
-static void check_unused(THD *thd)
-{
-  TABLE *entry;
-  TABLE_SHARE *share;
-  TDC_iterator tdc_it;
-  DBUG_ENTER("check_unused");
-
-  tdc_it.init();
-  mysql_mutex_lock(&LOCK_open);
-  while ((share= tdc_it.next()))
-  {
-    TABLE_SHARE::TABLE_list::Iterator it(share->tdc.free_tables);
-    while ((entry= it++))
-    {
-      /*
-        We must not have TABLEs in the free list that have their file closed.
-      */
-      DBUG_ASSERT(entry->db_stat && entry->file);
-      /* Merge children should be detached from a merge parent */
-      if (entry->in_use)
-      {
-        DBUG_PRINT("error",("Used table is in share's list of unused tables")); /* purecov: inspected */
-      }
-      /* extra() may assume that in_use is set */
-      entry->in_use= thd;
-      DBUG_ASSERT(!thd || !entry->file->extra(HA_EXTRA_IS_ATTACHED_CHILDREN));
-      entry->in_use= 0;
-    }
-  }
-  mysql_mutex_unlock(&LOCK_open);
-  tdc_it.deinit();
-  DBUG_VOID_RETURN;
-}
-#else
-#define check_unused(A)
-#endif
-
-
 /**
   Add new TABLE object to table cache.
 
@@ -301,7 +261,6 @@ void tc_add_table(THD *thd, TABLE *table)
       mysql_mutex_unlock(&LOCK_open);
       intern_close_table(purge_table);
       mysql_rwlock_unlock(&LOCK_flush);
-      check_unused(thd);
     }
     else
       mysql_mutex_unlock(&LOCK_open);
@@ -317,7 +276,9 @@ void tc_add_table(THD *thd, TABLE *table)
   Acquired object cannot be evicted or acquired again.
 
   While locked:
-  - pop object from TABLE_SHARE::tdc.free_tables()
+  - pop object from TABLE_SHARE::tdc.free_tables
+
+  While unlocked:
   - mark object used by thd
 
   @return TABLE object, or NULL if no unused objects.
@@ -328,19 +289,18 @@ static TABLE *tc_acquire_table(THD *thd, TABLE_SHARE *share)
   TABLE *table;
 
   mysql_mutex_lock(&LOCK_open);
-  if (!(table= share->tdc.free_tables.pop_front()))
-  {
+  table= share->tdc.free_tables.pop_front();
   mysql_mutex_unlock(&LOCK_open);
-    return 0;
-  }
+
+  if (table)
+  {
     DBUG_ASSERT(!table->in_use);
     table->in_use= thd;
-  mysql_mutex_unlock(&LOCK_open);
-
     /* The ex-unused table must be fully functional. */
     DBUG_ASSERT(table->db_stat && table->file);
     /* The children must be detached from the table. */
-  DBUG_ASSERT(! table->file->extra(HA_EXTRA_IS_ATTACHED_CHILDREN));
+    DBUG_ASSERT(!table->file->extra(HA_EXTRA_IS_ATTACHED_CHILDREN));
+  }
   return table;
 }
 
@@ -353,12 +313,12 @@ static TABLE *tc_acquire_table(THD *thd, TABLE_SHARE *share)
   Released object may be evicted or acquired again.
 
   While locked:
-  - mark object not in use by any thread
   - if object is marked for purge, decrement tc_count
   - add object to TABLE_SHARE::tdc.free_tables
   - evict LRU object from table cache if we reached threshold
 
   While unlocked:
+  - mark object not in use by any thread
   - free evicted/purged object
 
   @note Another thread may mark share for purge any moment (even
@@ -373,33 +333,37 @@ static TABLE *tc_acquire_table(THD *thd, TABLE_SHARE *share)
 
 bool tc_release_table(TABLE *table)
 {
-  THD *thd __attribute__((unused))= table->in_use;
   DBUG_ASSERT(table->in_use);
   DBUG_ASSERT(table->file);
 
   if (table->needs_reopen() || tc_records() > tc_size)
   {
     mysql_mutex_lock(&LOCK_open);
-    table->in_use= 0;
     goto purge;
   }
 
   table->tc_time= my_interval_timer();
 
   mysql_mutex_lock(&LOCK_open);
-  table->in_use= 0;
-  if (table->s->has_old_version())
+  if (table->s->tdc.flushed)
     goto purge;
+  /*
+    in_use doesn't really need protection of LOCK_open, 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().
+  */
+  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);
-  check_unused(thd);
   return false;
 
 purge:
   tc_remove_table(table);
   mysql_rwlock_rdlock(&LOCK_flush);
   mysql_mutex_unlock(&LOCK_open);
+  table->in_use= 0;
   intern_close_table(table);
   mysql_rwlock_unlock(&LOCK_flush);
   return true;
@@ -607,7 +571,8 @@ void tdc_init_share(TABLE_SHARE *share)
   share->tdc.all_tables.empty();
   share->tdc.free_tables.empty();
   tdc_assign_new_table_id(share);
-  share->version= tdc_refresh_version();
+  share->tdc.version= tdc_refresh_version();
+  share->tdc.flushed= false;
   DBUG_VOID_RETURN;
 }
 
@@ -767,7 +732,6 @@ TABLE_SHARE *tdc_acquire_share(THD *thd, const char *db, const char *table_name,
     if ((*out_table= tc_acquire_table(thd, share)))
     {
       mysql_rwlock_unlock(&LOCK_tdc);
-      check_unused(thd);
       DBUG_ASSERT(!(flags & GTS_NOLOCK));
       DBUG_ASSERT(!share->error);
       DBUG_ASSERT(!share->is_view);
@@ -855,7 +819,7 @@ void tdc_release_share(TABLE_SHARE *share)
   DBUG_PRINT("enter",
              ("share: 0x%lx  table: %s.%s  ref_count: %u  version: %lu",
               (ulong) share, share->db.str, share->table_name.str,
-              share->tdc.ref_count, share->version));
+              share->tdc.ref_count, share->tdc.version));
   DBUG_ASSERT(share->tdc.ref_count);
 
   if (share->tdc.ref_count > 1)
@@ -868,7 +832,7 @@ void tdc_release_share(TABLE_SHARE *share)
 
   mysql_mutex_lock(&LOCK_unused_shares);
   mysql_mutex_lock(&share->tdc.LOCK_table_share);
-  if (share->has_old_version())
+  if (share->tdc.flushed)
   {
     mysql_mutex_unlock(&share->tdc.LOCK_table_share);
     mysql_mutex_unlock(&LOCK_unused_shares);
@@ -988,17 +952,6 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
     I_P_List <TABLE, TABLE_share> purge_tables;
 
     mysql_mutex_lock(&LOCK_open);
-    if (kill_delayed_threads)
-      kill_delayed_threads_for_table(share);
-
-#ifndef DBUG_OFF
-    if (remove_type == TDC_RT_REMOVE_NOT_OWN)
-    {
-      TABLE_SHARE::All_share_tables_list::Iterator it2(share->tdc.all_tables);
-      while ((table= it2++))
-        DBUG_ASSERT(!table->in_use || table->in_use == thd);
-    }
-#endif
     /*
       Set share's version to zero in order to ensure that it gets
       automatically deleted once it is no longer referenced.
@@ -1008,13 +961,25 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
       shares.
     */
     if (remove_type != TDC_RT_REMOVE_NOT_OWN_KEEP_SHARE)
-      share->version= 0;
+      share->tdc.flushed= true;
 
     while ((table= share->tdc.free_tables.pop_front()))
     {
       tc_remove_table(table);
       purge_tables.push_front(table);
     }
+    if (kill_delayed_threads)
+      kill_delayed_threads_for_table(share);
+
+#ifndef DBUG_OFF
+    if (remove_type == TDC_RT_REMOVE_NOT_OWN)
+    {
+      TABLE_SHARE::All_share_tables_list::Iterator it(share->tdc.all_tables);
+      while ((table= it++))
+        DBUG_ASSERT(table->in_use == thd);
+    }
+#endif
+
     mysql_rwlock_rdlock(&LOCK_flush);
     mysql_mutex_unlock(&LOCK_open);
 
@@ -1022,7 +987,6 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
       intern_close_table(table);
     mysql_rwlock_unlock(&LOCK_flush);
 
-    check_unused(thd);
     DBUG_ASSERT(share->tdc.all_tables.is_empty() || remove_type != TDC_RT_REMOVE_ALL);
     tdc_release_share(share);
 
@@ -1052,14 +1016,15 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
 */
 
 int tdc_wait_for_old_version(THD *thd, const char *db, const char *table_name,
-                             ulong wait_timeout, uint deadlock_weight)
+                             ulong wait_timeout, uint deadlock_weight,
+                             ulong refresh_version)
 {
   TABLE_SHARE *share;
   int res= FALSE;
 
   if ((share= tdc_lock_share(db, table_name)))
   {
-    if (share->has_old_version())
+    if (share->tdc.flushed && refresh_version > share->tdc.version)
     {
       struct timespec abstime;
       set_timespec(abstime, wait_timeout);
@@ -1081,16 +1046,13 @@ ulong tdc_refresh_version(void)
 }
 
 
-void tdc_increment_refresh_version(void)
+ulong tdc_increment_refresh_version(void)
 {
   my_atomic_rwlock_wrlock(&LOCK_tdc_atomics);
-#ifndef DBUG_OFF
   ulong v= my_atomic_add64(&tdc_version, 1);
-#else
-  my_atomic_add64(&tdc_version, 1);
-#endif
   my_atomic_rwlock_wrunlock(&LOCK_tdc_atomics);
   DBUG_PRINT("tcache", ("incremented global refresh_version to: %lu", v));
+  return v + 1;
 }
 
 

sql/table_cache.h

@@ -47,13 +47,14 @@ extern bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
                              bool kill_delayed_threads);
 extern int tdc_wait_for_old_version(THD *thd, const char *db,
                                     const char *table_name,
-                                    ulong wait_timeout, uint deadlock_weight);
+                                    ulong wait_timeout, uint deadlock_weight,
+                                    ulong refresh_version= ULONG_MAX);
 extern ulong tdc_refresh_version(void);
-extern void tdc_increment_refresh_version(void);
+extern ulong tdc_increment_refresh_version(void);
 extern void tdc_assign_new_table_id(TABLE_SHARE *share);
 
 extern uint tc_records(void);
-extern void tc_purge(void);
+extern void tc_purge(bool mark_flushed= false);
 extern void tc_add_table(THD *thd, TABLE *table);
 extern bool tc_release_table(TABLE *table);