Making FLUSH TABLES WITH READ LOCK killable while it's waiting for running...

Making FLUSH TABLES WITH READ LOCK killable while it's waiting for running commits to finish. Normally this step is not long but it's still nice to be killable
(especially in case of bug like BUG#6732 "FLUSH TABLES WITH READ LOCK + COMMIT makes next FLUSH...LOCK hang forever").

mysql-test/r/flush_read_lock_kill.result

+drop table if exists t1;
+create table t1 (kill_id int);
+insert into t1 values(connection_id());
+ flush tables with read lock;
+select ((@id := kill_id) - kill_id) from t1;
+((@id := kill_id) - kill_id)
+0
+kill connection @id;
+drop table t1;

mysql-test/t/flush_read_lock_kill-master.opt

+--debug=d,make_global_read_lock_block_commit_loop

mysql-test/t/flush_read_lock_kill.test

+# Let's see if FLUSH TABLES WITH READ LOCK can be killed when waiting
+# for running commits to finish (in the past it could not)
+# This will not be a meaningful test on non-debug servers so will be
+# skipped.
+# If running mysql-test-run --debug, the --debug added by
+# mysql-test-run to the mysqld command line will override the one of
+# -master.opt. But this test is designed to still pass then (though it
+# won't test anything interesting).
+
+-- source include/have_debug.inc
+
+connect (con1,localhost,root,,);
+connect (con2,localhost,root,,);
+connection con1;
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+create table t1 (kill_id int);
+insert into t1 values(connection_id());
+
+# Thanks to the parameter we passed to --debug, this FLUSH will
+# block on a debug build running with our --debug=make_global... It
+# will block until killed. In other cases (non-debug build or other
+# --debug) it will succeed immediately
+
+connection con1;
+send flush tables with read lock;
+
+# kill con1
+connection con2;
+select ((@id := kill_id) - kill_id) from t1; 
+
+--sleep 2; # leave time for FLUSH to block
+kill connection @id;
+
+connection con1;
+# On debug builds it will be error 1053 (killed); on non-debug, or
+# debug build running without our --debug=make_global..., will be
+# error 0 (no error). The only important thing to test is that on
+# debug builds with our --debug=make_global... we don't hang forever.
+--error 0,1053
+reap;
+
+connection con2;
+drop table t1;

sql/lock.cc

@@ -840,19 +840,33 @@ void start_waiting_global_read_lock(THD *thd)
 }
 
 
-void make_global_read_lock_block_commit(THD *thd)
+bool make_global_read_lock_block_commit(THD *thd)
 {
+  bool error;
+  const char *old_message;
+  DBUG_ENTER("make_global_read_lock_block_commit");
   /*
     If we didn't succeed lock_global_read_lock(), or if we already suceeded
     make_global_read_lock_block_commit(), do nothing.
   */
   if (thd->global_read_lock != GOT_GLOBAL_READ_LOCK)
-    return;
+    DBUG_RETURN(1);
   pthread_mutex_lock(&LOCK_open);
   /* increment this BEFORE waiting on cond (otherwise race cond) */
   global_read_lock_blocks_commit++;
-  while (protect_against_global_read_lock)
+  /* For testing we set up some blocking, to see if we can be killed */
+  DBUG_EXECUTE_IF("make_global_read_lock_block_commit_loop",
+                  protect_against_global_read_lock++;);
+  old_message= thd->enter_cond(&COND_refresh, &LOCK_open,
+                               "Waiting for all running commits to finish");
+  while (protect_against_global_read_lock && !thd->killed)
     pthread_cond_wait(&COND_refresh, &LOCK_open);
-  pthread_mutex_unlock(&LOCK_open);
-  thd->global_read_lock= MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT;
+  DBUG_EXECUTE_IF("make_global_read_lock_block_commit_loop",
+                  protect_against_global_read_lock--;);
+  if (error= thd->killed)
+    global_read_lock_blocks_commit--; // undo what we did
+  else
+    thd->global_read_lock= MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT;
+  thd->exit_cond(old_message);
+  DBUG_RETURN(error);
 }

sql/mysql_priv.h

@@ -1096,7 +1096,7 @@ void unlock_global_read_lock(THD *thd);
 bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
                               bool is_not_commit);
 void start_waiting_global_read_lock(THD *thd);
-void make_global_read_lock_block_commit(THD *thd);
+bool make_global_read_lock_block_commit(THD *thd);
 
 /* Lock based on name */
 int lock_and_wait_for_table_name(THD *thd, TABLE_LIST *table_list);

sql/sql_parse.cc

@@ -5746,7 +5746,12 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
 	return 1;
       result=close_cached_tables(thd,(options & REFRESH_FAST) ? 0 : 1,
                                  tables);
-      make_global_read_lock_block_commit(thd);
+      if (make_global_read_lock_block_commit(thd))
+      {
+        /* Don't leave things in a half-locked state */
+        unlock_global_read_lock(thd);
+        return 1;
+      }
     }
     else
       result=close_cached_tables(thd,(options & REFRESH_FAST) ? 0 : 1, tables);