global read lock code now uses a dedicated mutex

(otherwise a deadlock when ALTER writes to
binlog holding LOCK_open, it causes binlog rotation,
binlog waits for prepared transactions to commit, and commit
needs LOCK_open to check for global read lock)

mysql-test/r/flush.result

@@ -8,7 +8,7 @@ n
 3
 flush tables with read lock;
 drop table t2;
-ERROR HY000: Table 't2' was locked with a READ lock and can't be updated
+ERROR HY000: Can't execute the query because you have a conflicting read lock
  drop table t2;
 unlock tables;
 create database mysqltest;

mysql-test/t/flush.test

@@ -37,7 +37,7 @@ connection con1;
 select * from t1;
 connection con2;
 flush tables with read lock;
---error 1099;
+--error 1223
 drop table t2;
 connection con1;
 send drop table t2;

sql/lock.cc

@@ -800,8 +800,8 @@ bool lock_global_read_lock(THD *thd)
 
   if (!thd->global_read_lock)
   {
-    (void) pthread_mutex_lock(&LOCK_open);
-    const char *old_message=thd->enter_cond(&COND_refresh, &LOCK_open,
+    (void) pthread_mutex_lock(&LOCK_global_read_lock);
+    const char *old_message=thd->enter_cond(&COND_refresh, &LOCK_global_read_lock,
 					    "Waiting to get readlock");
     DBUG_PRINT("info",
 	       ("waiting_for: %d  protect_against: %d",
@@ -809,7 +809,7 @@ bool lock_global_read_lock(THD *thd)
 
     waiting_for_read_lock++;
     while (protect_against_global_read_lock && !thd->killed)
-      pthread_cond_wait(&COND_refresh, &LOCK_open);
+      pthread_cond_wait(&COND_refresh, &LOCK_global_read_lock);
     waiting_for_read_lock--;
     if (thd->killed)
     {
@@ -834,11 +834,11 @@ bool lock_global_read_lock(THD *thd)
 void unlock_global_read_lock(THD *thd)
 {
   uint tmp;
-  pthread_mutex_lock(&LOCK_open);
+  pthread_mutex_lock(&LOCK_global_read_lock);
   tmp= --global_read_lock;
   if (thd->global_read_lock == MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT)
     --global_read_lock_blocks_commit;
-  pthread_mutex_unlock(&LOCK_open);
+  pthread_mutex_unlock(&LOCK_global_read_lock);
   /* Send the signal outside the mutex to avoid a context switch */
   if (!tmp)
     pthread_cond_broadcast(&COND_refresh);
@@ -857,7 +857,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
   DBUG_ENTER("wait_if_global_read_lock");
 
   LINT_INIT(old_message);
-  (void) pthread_mutex_lock(&LOCK_open);
+  (void) pthread_mutex_lock(&LOCK_global_read_lock);
   if ((need_exit_cond= must_wait))
   {
     if (thd->global_read_lock)		// This thread had the read locks
@@ -865,7 +865,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
       if (is_not_commit)
         my_message(ER_CANT_UPDATE_WITH_READLOCK,
                    ER(ER_CANT_UPDATE_WITH_READLOCK), MYF(0));
-      (void) pthread_mutex_unlock(&LOCK_open);
+      (void) pthread_mutex_unlock(&LOCK_global_read_lock);
       /*
         We allow FLUSHer to COMMIT; we assume FLUSHer knows what it does.
         This allowance is needed to not break existing versions of innobackup
@@ -873,11 +873,11 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
       */
       DBUG_RETURN(is_not_commit);
     }
-    old_message=thd->enter_cond(&COND_refresh, &LOCK_open,
+    old_message=thd->enter_cond(&COND_refresh, &LOCK_global_read_lock,
 				"Waiting for release of readlock");
     while (must_wait && ! thd->killed &&
 	   (!abort_on_refresh || thd->version == refresh_version))
-      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
+      (void) pthread_cond_wait(&COND_refresh,&LOCK_global_read_lock);
     if (thd->killed)
       result=1;
   }
@@ -890,7 +890,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
   if (unlikely(need_exit_cond)) 
     thd->exit_cond(old_message);
   else
-    pthread_mutex_unlock(&LOCK_open);
+    pthread_mutex_unlock(&LOCK_global_read_lock);
   DBUG_RETURN(result);
 }
 
@@ -901,10 +901,10 @@ void start_waiting_global_read_lock(THD *thd)
   DBUG_ENTER("start_waiting_global_read_lock");
   if (unlikely(thd->global_read_lock))
     DBUG_VOID_RETURN;
-  (void) pthread_mutex_lock(&LOCK_open);
+  (void) pthread_mutex_lock(&LOCK_global_read_lock);
   tmp= (!--protect_against_global_read_lock &&
         (waiting_for_read_lock || global_read_lock_blocks_commit));
-  (void) pthread_mutex_unlock(&LOCK_open);
+  (void) pthread_mutex_unlock(&LOCK_global_read_lock);
   if (tmp)
     pthread_cond_broadcast(&COND_refresh);
   DBUG_VOID_RETURN;
@@ -922,16 +922,16 @@ bool make_global_read_lock_block_commit(THD *thd)
   */
   if (thd->global_read_lock != GOT_GLOBAL_READ_LOCK)
     DBUG_RETURN(1);
-  pthread_mutex_lock(&LOCK_open);
+  pthread_mutex_lock(&LOCK_global_read_lock);
   /* increment this BEFORE waiting on cond (otherwise race cond) */
   global_read_lock_blocks_commit++;
   /* 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,
+  old_message= thd->enter_cond(&COND_refresh, &LOCK_global_read_lock,
                                "Waiting for all running commits to finish");
   while (protect_against_global_read_lock && !thd->killed)
-    pthread_cond_wait(&COND_refresh, &LOCK_open);
+    pthread_cond_wait(&COND_refresh, &LOCK_global_read_lock);
   DBUG_EXECUTE_IF("make_global_read_lock_block_commit_loop",
                   protect_against_global_read_lock--;);
   if (error= thd->killed)

sql/mysql_priv.h

@@ -1099,7 +1099,7 @@ extern pthread_mutex_t LOCK_mysql_create_db,LOCK_Acl,LOCK_open,
        LOCK_thread_count,LOCK_mapped_file,LOCK_user_locks, LOCK_status,
        LOCK_error_log, LOCK_delayed_insert, LOCK_uuid_generator,
        LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
-       LOCK_slave_list, LOCK_active_mi, LOCK_manager,
+       LOCK_slave_list, LOCK_active_mi, LOCK_manager, LOCK_global_read_lock,
        LOCK_global_system_variables, LOCK_user_conn;
 extern rw_lock_t LOCK_grant, LOCK_sys_init_connect, LOCK_sys_init_slave;
 extern pthread_cond_t COND_refresh, COND_thread_count, COND_manager;

sql/mysqld.cc

@@ -426,7 +426,7 @@ SHOW_COMP_OPTION have_crypt, have_compress;
 pthread_key(MEM_ROOT**,THR_MALLOC);
 pthread_key(THD*, THR_THD);
 pthread_mutex_t LOCK_mysql_create_db, LOCK_Acl, LOCK_open, LOCK_thread_count,
-		LOCK_mapped_file, LOCK_status,
+		LOCK_mapped_file, LOCK_status, LOCK_global_read_lock,
 		LOCK_error_log, LOCK_uuid_generator,
 		LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create,
 		LOCK_crypt, LOCK_bytes_sent, LOCK_bytes_received,
@@ -1102,6 +1102,7 @@ static void clean_up_mutexes()
   (void) rwlock_destroy(&LOCK_sys_init_connect);
   (void) rwlock_destroy(&LOCK_sys_init_slave);
   (void) pthread_mutex_destroy(&LOCK_global_system_variables);
+  (void) pthread_mutex_destroy(&LOCK_global_read_lock);
   (void) pthread_cond_destroy(&COND_thread_count);
   (void) pthread_cond_destroy(&COND_refresh);
   (void) pthread_cond_destroy(&COND_thread_cache);
@@ -2594,6 +2595,7 @@ static int init_thread_environment()
   (void) pthread_mutex_init(&LOCK_user_conn, MY_MUTEX_INIT_FAST);
   (void) pthread_mutex_init(&LOCK_active_mi, MY_MUTEX_INIT_FAST);
   (void) pthread_mutex_init(&LOCK_global_system_variables, MY_MUTEX_INIT_FAST);
+  (void) pthread_mutex_init(&LOCK_global_read_lock, MY_MUTEX_INIT_FAST);
   (void) pthread_mutex_init(&LOCK_uuid_generator, MY_MUTEX_INIT_FAST);
   (void) my_rwlock_init(&LOCK_sys_init_connect, NULL);
   (void) my_rwlock_init(&LOCK_sys_init_slave, NULL);

sql/sql_table.cc

@@ -65,7 +65,7 @@ static int copy_data_between_tables(TABLE *from,TABLE *to,
 bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
                     my_bool drop_temporary)
 {
-  bool error= FALSE;
+  bool error= FALSE, need_start_waiters= FALSE;
   DBUG_ENTER("mysql_rm_table");
 
   /* mark for close and remove all cached entries */
@@ -74,23 +74,19 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   thd->mysys_var->current_cond= &COND_refresh;
   VOID(pthread_mutex_lock(&LOCK_open));
 
-  if (!drop_temporary && global_read_lock)
+  if (!drop_temporary)
   {
-    if (thd->global_read_lock)
+    if ((error= wait_if_global_read_lock(thd, 0, 1)))
     {
       my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), tables->table_name);
-      error= TRUE;
       goto err;
     }
-    while (global_read_lock && ! thd->killed)
-    {
-      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
-    }
-
+    else
+      need_start_waiters= TRUE;
   }
   error= mysql_rm_table_part2(thd, tables, if_exists, drop_temporary, 0, 0);
 
- err:
+err:
   pthread_mutex_unlock(&LOCK_open);
 
   pthread_mutex_lock(&thd->mysys_var->mutex);
@@ -98,6 +94,9 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   thd->mysys_var->current_cond= 0;
   pthread_mutex_unlock(&thd->mysys_var->mutex);
 
+  if (need_start_waiters)
+    start_waiting_global_read_lock(thd);
+
   if (error)
     DBUG_RETURN(TRUE);
   send_ok(thd);
@@ -114,7 +113,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
     tables		List of tables to delete
     if_exists		If 1, don't give error if one table doesn't exists
     dont_log_query	Don't write query to log files. This will also not
-			generate warnings if the handler files doesn't exists  
+                        generate warnings if the handler files doesn't exists
 
  NOTES
    Works like documented in mysql_rm_table(), but don't check