Optimize flush tables with read lock (FTWRL) to not wait for select's

Part of MDEV-5336 Implement LOCK FOR BACKUP

The idea is that instead of waiting in close_cached_tables() for all
tables to be closed, we instead call flush_tables() that does:
- Flush not used objects in table cache to free memory
- Collect all tables that are open
- Call HA_EXTRA_FLUSH on the objects, to get them into "closed state"
- Added HA_EXTRA_FLUSH support to archive and CSV
- Added multi-user protection to HA_EXTRA_FLUSH in MyISAM and Aria

The benefit compared to old code is:
- FTWRL doesn't have to wait for long running read operations or
  open HANDLER's

9 files changed, 176 insertions, 31 deletions

diff --git a/sql/lock.cc b/sql/lock.cc
index 5420e9f42b5..59563487822 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -77,6 +77,7 @@
 #include "sql_base.h"                       // close_tables_for_reopen
 #include "sql_parse.h"                     // is_log_table_write_query
 #include "sql_acl.h"                       // SUPER_ACL
+#include "sql_handler.h"
 #include <hash.h>
 #include "wsrep_mysqld.h"
 
@@ -1015,6 +1016,8 @@ bool Global_read_lock::lock_global_read_lock(THD *thd)
   {
     MDL_request mdl_request;
 
+    mysql_ha_cleanup_no_free(thd);
+
     DBUG_ASSERT(! thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
                                                  MDL_SHARED));
     mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED, MDL_EXPLICIT);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 8dc68c32a0e..700175f14d6 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -349,6 +349,34 @@ static my_bool close_cached_tables_callback(TDC_element *element,
 }
 
 
+/**
+   Close all tables that are not in use in table definition cache
+
+   @param purge_flag  Argument for tc_purge. true if we should force all
+                      shares to be deleted. false if it's enough to just
+                      evict those that are not in use.
+*/
+
+void purge_tables(bool purge_flag)
+{
+  /*
+    Force close of all open tables.
+
+    Note that code in TABLE_SHARE::wait_for_old_version() assumes that
+    incrementing of refresh_version is followed by purge of unused table
+    shares.
+  */
+  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(purge_flag);
+  /* Free table shares which were not freed implicitly by loop above. */
+  tdc_purge(true);
+}
+
+
 bool close_cached_tables(THD *thd, TABLE_LIST *tables,
                          bool wait_for_refresh, ulong timeout)
 {
@@ -361,23 +389,7 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
   refresh_version= tdc_increment_refresh_version();
 
   if (!tables)
-  {
-    /*
-      Force close of all open tables.
-
-      Note that code in TABLE_SHARE::wait_for_old_version() assumes that
-      incrementing of refresh_version is followed by purge of unused table
-      shares.
-    */
-    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(true);
-    /* Free table shares which were not freed implicitly by loop above. */
-    tdc_purge(true);
-  }
+    purge_tables(true);
   else
   {
     bool found=0;
@@ -502,6 +514,128 @@ err_with_reopen:
 
 
 /**
+  Collect all shares that has open tables
+*/
+
+struct tc_collect_arg
+{
+  DYNAMIC_ARRAY shares;
+};
+
+static my_bool tc_collect_used_shares(TDC_element *element,
+                                      tc_collect_arg *arg)
+{
+  my_bool result= FALSE;
+
+  DYNAMIC_ARRAY *shares= &arg->shares;
+  mysql_mutex_lock(&element->LOCK_table_share);
+  if (element->ref_count > 0 && !element->share->is_view)
+  {
+    DBUG_ASSERT(element->share);
+    element->ref_count++;                       // Protect against delete
+    if (push_dynamic(shares,(uchar*) &element->share))
+      result= TRUE;
+  }
+  mysql_mutex_unlock(&element->LOCK_table_share);
+  return result;
+}
+
+
+/**
+   Flush cached table as part of global read lock
+
+   @param thd
+   @param flag   What type of tables should be flushed
+
+   @return 0  ok
+   @return 1  error
+
+   After we get the list of table shares, we will call flush on all
+   possible tables, even if some flush fails.
+*/
+
+bool flush_tables(THD *thd)
+{
+  bool result= TRUE;
+  uint open_errors= 0;
+  tc_collect_arg collect_arg;
+  TABLE *tmp_table;
+  DBUG_ENTER("flush_tables");
+
+  purge_tables(false);  /* Flush unused tables and shares */
+
+  /*
+    Loop over all shares and collect shares that have open tables
+    TODO:
+    Optimize this to only collect shares that have been used for
+    write after last time all tables was closed.
+  */
+
+  if (!(tmp_table= (TABLE*) my_malloc(sizeof(*tmp_table),
+                                      MYF(MY_WME | MY_THREAD_SPECIFIC))))
+    DBUG_RETURN(1);
+
+  my_init_dynamic_array(&collect_arg.shares, sizeof(TABLE_SHARE*), 100, 100,
+                        MYF(0));
+  if (tdc_iterate(thd, (my_hash_walk_action) tc_collect_used_shares,
+                  &collect_arg, true))
+  {
+    /* Release already collected shares */
+    for (uint i= 0 ; i < collect_arg.shares.elements ; i++)
+    {
+      TABLE_SHARE *share= *dynamic_element(&collect_arg.shares, i,
+                                           TABLE_SHARE**);
+      tdc_release_share(share);
+    }
+    goto err;
+  }
+
+  /* Call HA_EXTRA_FLUSH on all found shares */
+  for (uint i= 0 ; i < collect_arg.shares.elements ; i++)
+  {
+    TABLE_SHARE *share= *dynamic_element(&collect_arg.shares, i,
+                                         TABLE_SHARE**);
+    TABLE *table= tc_acquire_table(thd, share->tdc);
+    if (table)
+    {
+      (void) table->file->extra(HA_EXTRA_FLUSH);
+      tc_release_table(table);
+    }
+    else
+    {
+      /*
+        HA_OPEN_FOR_ALTER is used to allow us to open the table even if
+        TABLE_SHARE::incompatible_version is set.
+      */
+      if (!open_table_from_share(thd, share, &empty_clex_str,
+                                 HA_OPEN_KEYFILE, 0,
+                                 HA_OPEN_FOR_ALTER,
+                                 tmp_table, FALSE,
+                                 NULL))
+      {
+        (void) tmp_table->file->extra(HA_EXTRA_FLUSH);
+        /*
+          We don't put the table into the TDC as the table was not fully
+          opened (we didn't open triggers)
+        */
+        closefrm(tmp_table);
+      }
+      else
+        open_errors++;
+    }
+    tdc_release_share(share);
+  }
+
+  result= open_errors ? TRUE : FALSE;
+  DBUG_PRINT("note", ("open_errors: %u", open_errors));
+err:
+  my_free(tmp_table);
+  delete_dynamic(&collect_arg.shares);
+  DBUG_RETURN(result);
+}
+
+
+/**
   Close all tables which match specified connection string or
   if specified string is NULL, then any table with a connection string.
 */
diff --git a/sql/sql_base.h b/sql/sql_base.h
index bdadb420383..47ca2229af5 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -290,6 +290,8 @@ void close_log_table(THD *thd, Open_tables_backup *backup);
 
 bool close_cached_tables(THD *thd, TABLE_LIST *tables,
                          bool wait_for_refresh, ulong timeout);
+void purge_tables(bool purge_flag);
+bool flush_tables(THD *thd);
 bool close_cached_connection_tables(THD *thd, LEX_CSTRING *connect_string);
 void close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
                                ha_extra_function extra,
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 72df8367dc7..08114b99757 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -1196,10 +1196,10 @@ void mysql_ha_flush(THD *thd)
   @note Broadcasts refresh if it closed a table with old version.
 */
 
-void mysql_ha_cleanup(THD *thd)
+void mysql_ha_cleanup_no_free(THD *thd)
 {
   SQL_HANDLER *hash_tables;
-  DBUG_ENTER("mysql_ha_cleanup");
+  DBUG_ENTER("mysql_ha_cleanup_no_free");
 
   for (uint i= 0; i < thd->handler_tables_hash.records; i++)
   {
@@ -1207,9 +1207,15 @@ void mysql_ha_cleanup(THD *thd)
     if (hash_tables->table)
       mysql_ha_close_table(hash_tables);
   }
+  DBUG_VOID_RETURN;
+}
 
-  my_hash_free(&thd->handler_tables_hash);
 
+void mysql_ha_cleanup(THD *thd)
+{
+  DBUG_ENTER("mysql_ha_cleanup");
+  mysql_ha_cleanup_no_free(thd);
+  my_hash_free(&thd->handler_tables_hash);
   DBUG_VOID_RETURN;
 }
 
diff --git a/sql/sql_handler.h b/sql/sql_handler.h
index 4c16f7e5c57..16063bb1f35 100644
--- a/sql/sql_handler.h
+++ b/sql/sql_handler.h
@@ -73,6 +73,7 @@ bool mysql_ha_read(THD *, TABLE_LIST *,enum enum_ha_read_modes, const char *,
 void mysql_ha_flush(THD *thd);
 void mysql_ha_flush_tables(THD *thd, TABLE_LIST *all_tables);
 void mysql_ha_rm_tables(THD *thd, TABLE_LIST *tables);
+void mysql_ha_cleanup_no_free(THD *thd);
 void mysql_ha_cleanup(THD *thd);
 void mysql_ha_set_explicit_lock_duration(THD *thd);
 void mysql_ha_rm_temporary_tables(THD *thd);
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 5193524ca05..6a2af72e351 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2131,6 +2131,8 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     DBUG_EXECUTE_IF("simulate_detached_thread_refresh", debug_simulate= TRUE;);
     if (debug_simulate)
     {
+      /* This code doesn't work under FTWRL */
+      DBUG_ASSERT(! (options & REFRESH_READ_LOCK));
       /*
         Simulate a reload without a attached thread session.
         Provides a environment similar to that of when the
diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc
index abdf9d76d15..961f31eb728 100644
--- a/sql/sql_reload.cc
+++ b/sql/sql_reload.cc
@@ -231,6 +231,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
   {
     if ((options & REFRESH_READ_LOCK) && thd)
     {
+      DBUG_ASSERT(!(options & REFRESH_FAST) && !tables);
       /*
         On the first hand we need write lock on the tables to be flushed,
         on the other hand we must not try to aspire a global read lock
@@ -249,9 +250,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
       tmp_write_to_binlog= 0;
       if (thd->global_read_lock.lock_global_read_lock(thd))
 	return 1;                               // Killed
-      if (close_cached_tables(thd, tables,
-                              ((options & REFRESH_FAST) ?  FALSE : TRUE),
-                              thd->variables.lock_wait_timeout))
+      if (flush_tables(thd))
       {
         /*
           NOTE: my_error() has been already called by reopen_tables() within
@@ -274,11 +273,9 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
         make_global_read_lock_block_commit(thd) above since they could have
         modified the tables too.
       */
-      if (WSREP(thd) &&
-          close_cached_tables(thd, tables, (options & REFRESH_FAST) ?
-                              FALSE : TRUE, TRUE))
-          result= 1;
-     }
+      if (WSREP(thd) && flush_tables(thd))
+        result= 1;
+    }
     else
     {
       if (thd && thd->locked_tables_mode)
diff --git a/sql/table_cache.cc b/sql/table_cache.cc
index 9afe3ca61d4..d997aeff9f8 100644
--- a/sql/table_cache.cc
+++ b/sql/table_cache.cc
@@ -406,7 +406,7 @@ void tc_add_table(THD *thd, TABLE *table)
   @return TABLE object, or NULL if no unused objects.
 */
 
-static TABLE *tc_acquire_table(THD *thd, TDC_element *element)
+TABLE *tc_acquire_table(THD *thd, TDC_element *element)
 {
   uint32 n_instances=
     my_atomic_load32_explicit((int32*) &tc_active_instances,
@@ -657,7 +657,7 @@ void tdc_start_shutdown(void)
     tdc_size= 0;
     tc_size= 0;
     /* Free all cached but unused TABLEs and TABLE_SHAREs. */
-    close_cached_tables(NULL, NULL, FALSE, LONG_TIMEOUT);
+    purge_tables(true);
   }
   DBUG_VOID_RETURN;
 }
diff --git a/sql/table_cache.h b/sql/table_cache.h
index b41665258c9..148edc84223 100644
--- a/sql/table_cache.h
+++ b/sql/table_cache.h
@@ -88,7 +88,6 @@ extern bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
                              const char *db, const char *table_name,
                              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,
@@ -102,6 +101,7 @@ extern uint tc_records(void);
 extern void tc_purge(bool mark_flushed= false);
 extern void tc_add_table(THD *thd, TABLE *table);
 extern void tc_release_table(TABLE *table);
+extern TABLE *tc_acquire_table(THD *thd, TDC_element *element);
 
 /**
   Create a table cache key for non-temporary table.