bk commit into 5.1 tree (davi:1.2681) BUG#32395

[Davi Arnaut]

Below is the list of changes that have just been committed into a local
5.1 repository of davi. When davi does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet@stripped, 2007-12-11 11:21:15-02:00, davi@stripped +12 -0
  Bug#32395 Alter table under a impending global read lock causes a server crash
  
  The problem is that some DDS statements when under LOCK TABLES
  need to momentarily drop the lock, reopen the table and grab
  the write lock again (using reopen_tables). When grabbing the
  lock again, reopen_tables doesn't pass a flag to mysql_lock_tables
  in order to ignore the impending global read lock, which causes
  a assertion because LOCK_open is being hold. Also dropping the
  lock must not signal the flush thread that the table has been
  relinquished (related to the locking/flushing protocol).
  
  The solution is to correct the way the table is reopenned and
  the locks grabbed. When reopening the table and under LOCK
  TABLES, the table version should be set to 0 so other threads
  have to wait for the table. When grabbing the lock, any other
  flush should be ignored because it's theoretically a atomic
  operation.

  mysql-test/r/not_embedded_server.result@stripped, 2007-12-11 11:21:11-02:00, davi@stripped
+29 -0
    Add test case result for Bug#32395

  mysql-test/r/trigger_notembedded.result@stripped, 2007-12-11 11:21:11-02:00, davi@stripped
+14 -0
    Add test case result for Bug#32395

  mysql-test/t/lock_multi.test@stripped, 2007-12-11 11:21:11-02:00, davi@stripped +1 -1
    Update test case state.

  mysql-test/t/not_embedded_server.test@stripped, 2007-12-11 11:21:11-02:00, davi@stripped +95
-0
    Add test case for Bug#32395

  mysql-test/t/trigger_notembedded.test@stripped, 2007-12-11 11:21:11-02:00, davi@stripped +3
-3
    Enable test case for Bug#32395

  sql/ha_ndbcluster.cc@stripped, 2007-12-11 11:21:11-02:00, davi@stripped +1 -1
    Update close_cached_tables usage.

  sql/ha_ndbcluster_binlog.cc@stripped, 2007-12-11 11:21:12-02:00, davi@stripped +6 -6
    Update close_cached_tables usage.

  sql/mysql_priv.h@stripped, 2007-12-11 11:21:12-02:00, davi@stripped +2 -1
    Update close_cache_tables prototype.

  sql/set_var.cc@stripped, 2007-12-11 11:21:12-02:00, davi@stripped +1 -1
    Update close_cached_tables usage.

  sql/sql_base.cc@stripped, 2007-12-11 11:21:12-02:00, davi@stripped +43 -23
    Preserve old version for write locked tables and ignore
    pending flushes and update close_cache_tables to take
    into account name locked tables.

  sql/sql_parse.cc@stripped, 2007-12-11 11:21:12-02:00, davi@stripped +4 -3
    Update close_cached_tables usage and pass flag so that
    name locked tables are waited for.

  sql/sql_table.cc@stripped, 2007-12-11 11:21:13-02:00, davi@stripped +9 -6
    Don't wait for the impending GRL if under LOCK TABLES.

diff -Nrup a/mysql-test/r/not_embedded_server.result
b/mysql-test/r/not_embedded_server.result
--- a/mysql-test/r/not_embedded_server.result	2006-09-06 09:32:43 -03:00
+++ b/mysql-test/r/not_embedded_server.result	2007-12-11 11:21:11 -02:00
@@ -3,3 +3,32 @@ execute stmt1;
 ID	USER	HOST	DB	COMMAND	TIME	STATE	INFO
 number	root	localhost	test	Query	time	preparing	SELECT * FROM
INFORMATION_SCHEMA.PROCESSLIST WHERE COMMAND!='Daemon'
 deallocate prepare stmt1;
+drop table if exists t1;
+create table t1 (i int);
+connection: default
+lock tables t1 write;
+connection: flush
+flush tables with read lock;;
+connection: default
+alter table t1 add column j int;
+connection: insert
+insert into t1 values (1,2);;
+connection: default
+unlock tables;
+connection: flush
+select * from t1;
+i	j
+unlock tables;
+select * from t1;
+i	j
+1	2
+drop table t1;
+drop table if exists t1;
+create table t1 (i int);
+connection: default
+lock tables t1 write;
+connection: flush
+flush tables with read lock;;
+connection: default
+flush tables;
+drop table t1;
diff -Nrup a/mysql-test/r/trigger_notembedded.result
b/mysql-test/r/trigger_notembedded.result
--- a/mysql-test/r/trigger_notembedded.result	2007-11-29 09:42:23 -02:00
+++ b/mysql-test/r/trigger_notembedded.result	2007-12-11 11:21:11 -02:00
@@ -448,4 +448,18 @@ DROP TABLE t1;
 DROP DATABASE mysqltest_db1;
 USE test;
 End of 5.0 tests.
+drop table if exists t1;
+create table t1 (i int);
+connection: default
+lock tables t1 write;
+connection: flush
+flush tables with read lock;;
+connection: default
+create trigger t1_bi before insert on t1 for each row begin end;
+unlock tables;
+connection: flush
+unlock tables;
+select * from t1;
+i
+drop table t1;
 End of 5.1 tests.
diff -Nrup a/mysql-test/t/lock_multi.test b/mysql-test/t/lock_multi.test
--- a/mysql-test/t/lock_multi.test	2007-11-28 12:33:39 -02:00
+++ b/mysql-test/t/lock_multi.test	2007-12-11 11:21:11 -02:00
@@ -150,7 +150,7 @@ send SELECT user.Select_priv FROM user, 
 connection locker;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = 
+  where state = "Waiting for table" and info = 
   "SELECT user.Select_priv FROM user, db WHERE user.user = db.user LIMIT 1";
 --source include/wait_condition.inc
 # Make test case independent from earlier grants.
diff -Nrup a/mysql-test/t/not_embedded_server.test b/mysql-test/t/not_embedded_server.test
--- a/mysql-test/t/not_embedded_server.test	2006-08-15 08:10:07 -03:00
+++ b/mysql-test/t/not_embedded_server.test	2007-12-11 11:21:11 -02:00
@@ -20,4 +20,99 @@ prepare stmt1 from ' SELECT * FROM INFOR
 execute stmt1;
 deallocate prepare stmt1;
 
+#
+# Bug#32395 Alter table under a impending global read lock causes a server crash
+#
+
+#
+# Test ALTER TABLE under LOCK TABLES and FLUSH TABLES WITH READ LOCK
+#
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+create table t1 (i int);
+connect (flush,localhost,root,,test,,);
+connection default;
+--echo connection: default
+lock tables t1 write;
+connection flush;
+--echo connection: flush
+--send flush tables with read lock;
+connection default;
+--echo connection: default
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = "Flushing tables";
+--source include/wait_condition.inc
+alter table t1 add column j int;
+connect (insert,localhost,root,,test,,);
+connection insert;
+--echo connection: insert
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = "Flushing tables";
+--source include/wait_condition.inc
+--send insert into t1 values (1,2);
+--echo connection: default
+connection default;
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = "Waiting for release of readlock";
+--source include/wait_condition.inc
+unlock tables;
+connection flush;
+--echo connection: flush
+--reap
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = "Waiting for release of readlock";
+--source include/wait_condition.inc
+select * from t1;
+unlock tables;
+connection insert;
+--reap
+connection default;
+select * from t1;
+drop table t1;
+disconnect flush;
+disconnect insert;
+
+#
+# Test that FLUSH TABLES under LOCK TABLES protects write locked tables
+# from a impending FLUSH TABLES WITH READ LOCK
+#
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+create table t1 (i int);
+connect (flush,localhost,root,,test,,);
+connection default;
+--echo connection: default
+lock tables t1 write;
+connection flush;
+--echo connection: flush
+--send flush tables with read lock;
+connection default;
+--echo connection: default
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = "Flushing tables";
+--source include/wait_condition.inc
+flush tables;
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = "Flushing tables";
+--source include/wait_condition.inc
+drop table t1;
+let $wait_condition=
+  select count(*) = 0 from information_schema.processlist
+  where state = "Flushing tables";
+--source include/wait_condition.inc
+connection flush;
+--reap
+connection default;
+disconnect flush;
+
 # End of 5.1 tests
diff -Nrup a/mysql-test/t/trigger_notembedded.test b/mysql-test/t/trigger_notembedded.test
--- a/mysql-test/t/trigger_notembedded.test	2007-11-29 09:42:23 -02:00
+++ b/mysql-test/t/trigger_notembedded.test	2007-12-11 11:21:11 -02:00
@@ -880,8 +880,9 @@ USE test;
 # Bug#23713 LOCK TABLES + CREATE TRIGGER + FLUSH TABLES WITH READ LOCK = deadlock
 #
 
-# Test temporarily disable due to Bug#32395
---disable_parsing
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
 create table t1 (i int);
 connect (flush,localhost,root,,test,,);
 connection default;
@@ -906,6 +907,5 @@ connection default;
 select * from t1;
 drop table t1;
 disconnect flush;
---enable_parsing
 
 --echo End of 5.1 tests.
diff -Nrup a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
--- a/sql/ha_ndbcluster.cc	2007-11-21 18:09:21 -02:00
+++ b/sql/ha_ndbcluster.cc	2007-12-11 11:21:11 -02:00
@@ -577,7 +577,7 @@ int ha_ndbcluster::ndb_err(NdbTransactio
     bzero((char*) &table_list,sizeof(table_list));
     table_list.db= m_dbname;
     table_list.alias= table_list.table_name= m_tabname;
-    close_cached_tables(thd, 0, &table_list);
+    close_cached_tables(thd, &table_list, FALSE, FALSE, FALSE);
     break;
   }
   default:
diff -Nrup a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
--- a/sql/ha_ndbcluster_binlog.cc	2007-11-16 11:06:27 -02:00
+++ b/sql/ha_ndbcluster_binlog.cc	2007-12-11 11:21:12 -02:00
@@ -883,7 +883,7 @@ int ndbcluster_setup_binlog_table_shares
     {
       if (ndb_extra_logging)
         sql_print_information("NDB Binlog: ndb tables writable");
-      close_cached_tables((THD*) 0, 0, (TABLE_LIST*) 0, TRUE);
+      close_cached_tables(NULL, NULL, TRUE, FALSE, FALSE);
     }
     pthread_mutex_unlock(&LOCK_open);
     /* Signal injector thread that all is setup */
@@ -1683,7 +1683,7 @@ ndb_handle_schema_change(THD *thd, Ndb *
       bzero((char*) &table_list,sizeof(table_list));
       table_list.db= (char *)dbname;
       table_list.alias= table_list.table_name= (char *)tabname;
-      close_cached_tables(thd, 0, &table_list, TRUE);
+      close_cached_tables(thd, &table_list, TRUE, FALSE, FALSE);
 
       if ((error= ndbcluster_binlog_open_table(thd, share,
                                                table_share, table, 1)))
@@ -1789,7 +1789,7 @@ ndb_handle_schema_change(THD *thd, Ndb *
     bzero((char*) &table_list,sizeof(table_list));
     table_list.db= (char *)dbname;
     table_list.alias= table_list.table_name= (char *)tabname;
-    close_cached_tables(thd, 0, &table_list);
+    close_cached_tables(thd, &table_list, FALSE, FALSE, FALSE);
     /* ndb_share reference create free */
     DBUG_PRINT("NDB_SHARE", ("%s create free  use_count: %u",
                              share->key, share->use_count));
@@ -1908,7 +1908,7 @@ ndb_binlog_thread_handle_schema_event(TH
             bzero((char*) &table_list,sizeof(table_list));
             table_list.db= schema->db;
             table_list.alias= table_list.table_name= schema->name;
-            close_cached_tables(thd, 0, &table_list, FALSE);
+            close_cached_tables(thd, &table_list, FALSE, FALSE, FALSE);
           }
           /* ndb_share reference temporary free */
           if (share)
@@ -2032,7 +2032,7 @@ ndb_binlog_thread_handle_schema_event(TH
       pthread_mutex_unlock(&ndb_schema_share_mutex);
       /* end protect ndb_schema_share */
 
-      close_cached_tables((THD*) 0, 0, (TABLE_LIST*) 0, FALSE);
+      close_cached_tables(NULL, NULL, FALSE, FALSE, FALSE);
       // fall through
     case NDBEVENT::TE_ALTER:
       ndb_handle_schema_change(thd, ndb, pOp, tmp_share);
@@ -2189,7 +2189,7 @@ ndb_binlog_thread_handle_schema_event_po
           bzero((char*) &table_list,sizeof(table_list));
           table_list.db= schema->db;
           table_list.alias= table_list.table_name= schema->name;
-          close_cached_tables(thd, 0, &table_list, FALSE);
+          close_cached_tables(thd, &table_list, FALSE, FALSE, FALSE);
         }
         if (schema_type != SOT_ALTER_TABLE)
           break;
diff -Nrup a/sql/mysql_priv.h b/sql/mysql_priv.h
--- a/sql/mysql_priv.h	2007-11-29 09:42:23 -02:00
+++ b/sql/mysql_priv.h	2007-12-11 11:21:12 -02:00
@@ -1595,7 +1595,8 @@ TABLE *open_performance_schema_table(THD
                                      Open_tables_state *backup);
 void close_performance_schema_table(THD *thd, Open_tables_state *backup);
 
-bool close_cached_tables(THD *thd, bool wait_for_refresh, TABLE_LIST *tables, bool
have_lock = FALSE);
+bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
+                         bool wait_for_refresh, bool wait_for_locked);
 bool close_cached_connection_tables(THD *thd, bool wait_for_refresh,
                                     LEX_STRING *connect_string,
                                     bool have_lock = FALSE);
diff -Nrup a/sql/set_var.cc b/sql/set_var.cc
--- a/sql/set_var.cc	2007-11-14 11:28:19 -02:00
+++ b/sql/set_var.cc	2007-12-11 11:21:12 -02:00
@@ -3685,7 +3685,7 @@ bool sys_var_opt_readonly::update(THD *t
     can cause to wait on a read lock, it's required for the client application
     to unlock everything, and acceptable for the server to wait on all locks.
   */
-  if (result= close_cached_tables(thd, true, NULL, false))
+  if (result= close_cached_tables(thd, NULL, FALSE, TRUE, FALSE))
     goto end_with_read_lock;
 
   if (result= make_global_read_lock_block_commit(thd))
diff -Nrup a/sql/sql_base.cc b/sql/sql_base.cc
--- a/sql/sql_base.cc	2007-11-29 09:42:23 -02:00
+++ b/sql/sql_base.cc	2007-12-11 11:21:12 -02:00
@@ -130,7 +130,7 @@ void table_cache_free(void)
   DBUG_ENTER("table_cache_free");
   if (table_def_inited)
   {
-    close_cached_tables((THD*) 0,0,(TABLE_LIST*) 0);
+    close_cached_tables(NULL, NULL, FALSE, FALSE, FALSE);
     if (!open_cache.records)			// Safety first
       hash_free(&open_cache);
   }
@@ -885,16 +885,22 @@ void free_io_cache(TABLE *table)
 /*
   Close all tables which aren't in use by any thread
 
-  THD can be NULL, but then if_wait_for_refresh must be FALSE
-  and tables must be NULL.
+  @param thd Thread context
+  @param tables List of tables to remove from the cache
+  @param have_lock If LOCK_open is locked
+  @param wait_for_refresh Wait for a impending flush
+  @param wait_for_locked Wait for locked table name
+
+  @remark THD can be NULL, but then wait_for_refresh must be FALSE
+          and tables must be NULL.
 */
 
-bool close_cached_tables(THD *thd, bool if_wait_for_refresh,
-			 TABLE_LIST *tables, bool have_lock)
+bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
+                         bool wait_for_refresh, bool wait_for_locked)
 {
   bool result=0;
   DBUG_ENTER("close_cached_tables");
-  DBUG_ASSERT(thd || (!if_wait_for_refresh && !tables));
+  DBUG_ASSERT(thd || (!wait_for_refresh && !tables));
 
   if (!have_lock)
     VOID(pthread_mutex_lock(&LOCK_open));
@@ -918,7 +924,7 @@ bool close_cached_tables(THD *thd, bool 
     }
     DBUG_PRINT("tcache", ("incremented global refresh_version to: %lu",
                           refresh_version));
-    if (if_wait_for_refresh)
+    if (wait_for_refresh)
     {
       /*
         Other threads could wait in a loop in open_and_lock_tables(),
@@ -975,13 +981,13 @@ bool close_cached_tables(THD *thd, bool 
 	found=1;
     }
     if (!found)
-      if_wait_for_refresh=0;			// Nothing to wait for
+      wait_for_refresh=0;			// Nothing to wait for
   }
 #ifndef EMBEDDED_LIBRARY
   if (!tables)
     kill_delayed_threads();
 #endif
-  if (if_wait_for_refresh)
+  if (wait_for_refresh)
   {
     /*
       If there is any table that has a lower refresh_version, wait until
@@ -1004,6 +1010,9 @@ bool close_cached_tables(THD *thd, bool 
       for (uint idx=0 ; idx < open_cache.records ; idx++)
       {
 	TABLE *table=(TABLE*) hash_element(&open_cache,idx);
+        /* Avoid a self-deadlock. */
+        if (table->in_use == thd)
+          continue;
         /*
           Note that we wait here only for tables which are actually open, and
           not for placeholders with TABLE::open_placeholder set. Waiting for
@@ -1018,7 +1027,8 @@ bool close_cached_tables(THD *thd, bool 
           are employed by CREATE TABLE as in this case table simply does not
           exist yet.
         */
-	if (table->needs_reopen_or_name_lock() && table->db_stat)
+	if (table->needs_reopen_or_name_lock() && (table->db_stat ||
+            (table->open_placeholder && wait_for_locked)))
 	{
 	  found=1;
           DBUG_PRINT("signal", ("Waiting for COND_refresh"));
@@ -1037,11 +1047,18 @@ bool close_cached_tables(THD *thd, bool 
     thd->in_lock_tables=0;
     /* Set version for table */
     for (TABLE *table=thd->open_tables; table ; table= table->next)
-      table->s->version= refresh_version;
+    {
+      /*
+        Preserve the version (0) of write locked tables so that a impending
+        global read lock won't sneak in.
+      */
+      if (table->reginfo.lock_type < TL_WRITE_LOW_PRIORITY)
+        table->s->version= refresh_version;
+    }
   }
   if (!have_lock)
     VOID(pthread_mutex_unlock(&LOCK_open));
-  if (if_wait_for_refresh)
+  if (wait_for_refresh)
   {
     pthread_mutex_lock(&thd->mysys_var->mutex);
     thd->mysys_var->current_mutex= 0;
@@ -1068,10 +1085,10 @@ bool close_cached_connection_tables(THD 
   DBUG_ASSERT(thd);
 
   bzero(&tmp, sizeof(TABLE_LIST));
-  
+
   if (!have_lock)
     VOID(pthread_mutex_lock(&LOCK_open));
-  
+
   for (idx= 0; idx < table_def_cache.records; idx++)
   {
     TABLE_SHARE *share= (TABLE_SHARE *) hash_element(&table_def_cache, idx);
@@ -1100,11 +1117,11 @@ bool close_cached_connection_tables(THD 
   }
 
   if (tables)
-    result= close_cached_tables(thd, FALSE, tables, TRUE);
-  
+    result= close_cached_tables(thd, tables, TRUE, FALSE, FALSE);
+
   if (!have_lock)
     VOID(pthread_mutex_unlock(&LOCK_open));
-  
+
   if (if_wait_for_refresh)
   {
     pthread_mutex_lock(&thd->mysys_var->mutex);
@@ -2204,7 +2221,7 @@ void wait_for_condition(THD *thd, pthrea
   current thread.
 
   @param thd current thread context
-  @param tables able list containing one table to open.
+  @param tables table list containing one table to open.
 
   @return FALSE on success, TRUE otherwise.
 */
@@ -3272,8 +3289,7 @@ static bool reattach_merge(THD *thd, TAB
 
     @param thd         Thread context
     @param get_locks   Should we get locks after reopening tables ?
-    @param in_refresh  Are we in FLUSH TABLES ? TODO: It seems that
-                       we can remove this parameter.
+    @param mark_old    Mark share as old
 
     @note Since this function can't properly handle prelocking and
           create placeholders it should be used in very special
@@ -3287,13 +3303,17 @@ static bool reattach_merge(THD *thd, TAB
     @return FALSE in case of success, TRUE - otherwise.
 */
 
-bool reopen_tables(THD *thd,bool get_locks,bool in_refresh)
+bool reopen_tables(THD *thd, bool get_locks, bool mark_old)
 {
   TABLE *table,*next,**prev;
   TABLE **tables,**tables_ptr;			// For locks
   TABLE *err_tables= NULL;
   bool error=0, not_used;
   bool merge_table_found= FALSE;
+  const uint flags= MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN |
+                    MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK |
+                    MYSQL_LOCK_IGNORE_FLUSH;
+
   DBUG_ENTER("reopen_tables");
 
   if (!thd->open_tables)
@@ -3354,7 +3374,7 @@ bool reopen_tables(THD *thd,bool get_loc
       /* Do not handle locks of MERGE children. */
       if (get_locks && !db_stat && !table->parent)
 	*tables_ptr++= table;			// need new lock on this
-      if (in_refresh)
+      if (mark_old)
       {
 	table->s->version=0;
 	table->open_placeholder= 0;
@@ -3387,7 +3407,7 @@ bool reopen_tables(THD *thd,bool get_loc
     */
     thd->some_tables_deleted=0;
     if ((lock= mysql_lock_tables(thd, tables, (uint) (tables_ptr - tables),
-                                 MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN, &not_used)))
+                                 flags, &not_used)))
     {
       thd->locked_tables=mysql_lock_merge(thd->locked_tables,lock);
     }
diff -Nrup a/sql/sql_parse.cc b/sql/sql_parse.cc
--- a/sql/sql_parse.cc	2007-12-10 13:43:23 -02:00
+++ b/sql/sql_parse.cc	2007-12-11 11:21:12 -02:00
@@ -6530,8 +6530,8 @@ bool reload_acl_and_cache(THD *thd, ulon
       tmp_write_to_binlog= 0;
       if (lock_global_read_lock(thd))
 	return 1;                               // Killed
-      result=close_cached_tables(thd,(options & REFRESH_FAST) ? 0 : 1,
-                                 tables);
+      result= close_cached_tables(thd, tables, FALSE, (options & REFRESH_FAST) ?
+                                  FALSE : TRUE, TRUE);
       if (make_global_read_lock_block_commit(thd)) // Killed
       {
         /* Don't leave things in a half-locked state */
@@ -6540,7 +6540,8 @@ bool reload_acl_and_cache(THD *thd, ulon
       }
     }
     else
-      result=close_cached_tables(thd,(options & REFRESH_FAST) ? 0 : 1, tables);
+      result= close_cached_tables(thd, tables, FALSE, (options & REFRESH_FAST) ?
+                                  FALSE : TRUE, FALSE);
     my_dbopt_cleanup();
   }
   if (options & REFRESH_HOSTS)
diff -Nrup a/sql/sql_table.cc b/sql/sql_table.cc
--- a/sql/sql_table.cc	2007-11-29 09:42:24 -02:00
+++ b/sql/sql_table.cc	2007-12-11 11:21:13 -02:00
@@ -1406,7 +1406,6 @@ void write_bin_log(THD *thd, bool clear_
   }
 }
 
-
 /*
  delete (drop) tables.
 
@@ -1422,7 +1421,8 @@ void write_bin_log(THD *thd, bool clear_
     If a table is in use, we will wait for all users to free the table
     before dropping it
 
-    Wait if global_read_lock (FLUSH TABLES WITH READ LOCK) is set.
+    Wait if global_read_lock (FLUSH TABLES WITH READ LOCK) is set, but
+    not if under LOCK TABLES.
 
   RETURN
     FALSE OK.  In this case ok packet is sent to user
@@ -1440,13 +1440,16 @@ bool mysql_rm_table(THD *thd,TABLE_LIST 
 
   if (!drop_temporary)
   {
-    if ((error= wait_if_global_read_lock(thd, 0, 1)))
+    /*
+      lock_table_name will error out if under LOCK TABLES and the tables
+      are not write locked.
+    */
+    if (!thd->locked_tables &&
+        !(need_start_waiters= !wait_if_global_read_lock(thd, 0, 1)))
     {
       my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), tables->table_name);
       DBUG_RETURN(TRUE);
     }
-    else
-      need_start_waiters= TRUE;
   }
 
   /*
@@ -6665,7 +6668,7 @@ view_err:
   if (thd->locked_tables && new_name == table_name && new_db == db)
   {
     thd->in_lock_tables= 1;
-    error= reopen_tables(thd, 1, 0);
+    error= reopen_tables(thd, 1, 1);
     thd->in_lock_tables= 0;
     if (error)
       goto err_with_placeholders;