Merge acurtis@bk-internal.mysql.com:/home/bk/mysql-5.1-engines

into  xiphis.org:/home/antony/work2/p2-bug20168.2

mysql-test/r/lock_multi.result

@@ -66,3 +66,11 @@ Select_priv
 N
 use test;
 use test;
+CREATE DATABASE mysqltest_1;
+FLUSH TABLES WITH READ LOCK;
+ DROP DATABASE mysqltest_1;
+DROP DATABASE mysqltest_1;
+ERROR HY000: Can't execute the query because you have a conflicting read lock
+UNLOCK TABLES;
+DROP DATABASE mysqltest_1;
+ERROR HY000: Can't drop database 'mysqltest_1'; database doesn't exist

mysql-test/t/lock_multi.test

@@ -158,3 +158,36 @@ use test;
 connection default;
 
 # End of 5.0 tests
+# Bug#19815 - CREATE/RENAME/DROP DATABASE can deadlock on a global read lock
+#
+connect (con1,localhost,root,,);
+connect (con2,localhost,root,,);
+#
+connection con1;
+CREATE DATABASE mysqltest_1;
+FLUSH TABLES WITH READ LOCK;
+#
+# With bug in place: acquire LOCK_mysql_create_table and
+# wait in wait_if_global_read_lock().
+connection con2;
+send DROP DATABASE mysqltest_1;
+--sleep 1
+#
+# With bug in place: try to acquire LOCK_mysql_create_table...
+# When fixed: Reject dropping db because of the read lock.
+connection con1;
+--error ER_CANT_UPDATE_WITH_READLOCK
+DROP DATABASE mysqltest_1;
+UNLOCK TABLES;
+#
+connection con2;
+reap;
+#
+connection default;
+disconnect con1;
+disconnect con2;
+# This must have been dropped by connection 2 already,
+# which waited until the global read lock was released.
+--error ER_DB_DROP_EXISTS
+DROP DATABASE mysqltest_1;
+

sql/sql_db.cc

@@ -538,16 +538,27 @@ bool mysql_create_db(THD *thd, char *db, HA_CREATE_INFO *create_info,
     my_error(ER_DB_CREATE_EXISTS, MYF(0), db);
     DBUG_RETURN(-1);
   }
-  
-  VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
 
-  /* do not create database if another thread is holding read lock */
+  /*
+    Do not create database if another thread is holding read lock.
+    Wait for global read lock before acquiring LOCK_mysql_create_db.
+    After wait_if_global_read_lock() we have protection against another
+    global read lock. If we would acquire LOCK_mysql_create_db first,
+    another thread could step in and get the global read lock before we
+    reach wait_if_global_read_lock(). If this thread tries the same as we
+    (admin a db), it would then go and wait on LOCK_mysql_create_db...
+    Furthermore wait_if_global_read_lock() checks if the current thread
+    has the global read lock and refuses the operation with
+    ER_CANT_UPDATE_WITH_READLOCK if applicable.
+  */
   if (wait_if_global_read_lock(thd, 0, 1))
   {
     error= -1;
     goto exit2;
   }
 
+  VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
+
   /* Check directory */
   path_len= build_table_filename(path, sizeof(path), db, "", "");
   path[path_len-1]= 0;                    // Remove last '/' from path
@@ -655,9 +666,9 @@ bool mysql_create_db(THD *thd, char *db, HA_CREATE_INFO *create_info,
   }
 
 exit:
+  VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
   start_waiting_global_read_lock(thd);
 exit2:
-  VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
   DBUG_RETURN(error);
 }
 
@@ -671,12 +682,23 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
   int error= 0;
   DBUG_ENTER("mysql_alter_db");
 
-  VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
-
-  /* do not alter database if another thread is holding read lock */
+  /*
+    Do not alter database if another thread is holding read lock.
+    Wait for global read lock before acquiring LOCK_mysql_create_db.
+    After wait_if_global_read_lock() we have protection against another
+    global read lock. If we would acquire LOCK_mysql_create_db first,
+    another thread could step in and get the global read lock before we
+    reach wait_if_global_read_lock(). If this thread tries the same as we
+    (admin a db), it would then go and wait on LOCK_mysql_create_db...
+    Furthermore wait_if_global_read_lock() checks if the current thread
+    has the global read lock and refuses the operation with
+    ER_CANT_UPDATE_WITH_READLOCK if applicable.
+  */
   if ((error=wait_if_global_read_lock(thd,0,1)))
     goto exit2;
 
+  VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
+
   /* 
      Recreate db options file: /dbpath/.db.opt
      We pass MY_DB_OPT_FILE as "extension" to avoid
@@ -721,9 +743,9 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
   send_ok(thd, result);
 
 exit:
+  VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
   start_waiting_global_read_lock(thd);
 exit2:
-  VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
   DBUG_RETURN(error);
 }
 
@@ -755,15 +777,26 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   TABLE_LIST* dropped_tables= 0;
   DBUG_ENTER("mysql_rm_db");
 
-  VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
-
-  /* do not drop database if another thread is holding read lock */
+  /*
+    Do not drop database if another thread is holding read lock.
+    Wait for global read lock before acquiring LOCK_mysql_create_db.
+    After wait_if_global_read_lock() we have protection against another
+    global read lock. If we would acquire LOCK_mysql_create_db first,
+    another thread could step in and get the global read lock before we
+    reach wait_if_global_read_lock(). If this thread tries the same as we
+    (admin a db), it would then go and wait on LOCK_mysql_create_db...
+    Furthermore wait_if_global_read_lock() checks if the current thread
+    has the global read lock and refuses the operation with
+    ER_CANT_UPDATE_WITH_READLOCK if applicable.
+  */
   if (wait_if_global_read_lock(thd, 0, 1))
   {
     error= -1;
     goto exit2;
   }
 
+  VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
+
   length= build_table_filename(path, sizeof(path), db, "", "");
   strmov(path+length, MY_DB_OPT_FILE);		// Append db option file name
   del_dbopt(path);				// Remove dboption hash entry
@@ -872,7 +905,6 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
 exit:
   (void)sp_drop_db_routines(thd, db); /* QQ Ignore errors for now  */
   error= Events::drop_schema_events(thd, db);
-  start_waiting_global_read_lock(thd);
   /*
     If this database was the client's selected database, we silently change the
     client's selected database to nothing (to have an empty SELECT DATABASE()
@@ -901,9 +933,9 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     thd->db= 0;
     thd->db_length= 0;
   }
-exit2:
   VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
-
+  start_waiting_global_read_lock(thd);
+exit2:
   DBUG_RETURN(error);
 }
 

storage/myisam/mi_check.c

@@ -453,25 +453,24 @@ int chk_key(MI_CHECK *param, register MI_INFO *info)
     if ((uint) share->base.auto_key -1 == key)
     {
       /* Check that auto_increment key is bigger than max key value */
-      ulonglong save_auto_value=info->s->state.auto_increment;
-      info->s->state.auto_increment=0;
+      ulonglong auto_increment;
       info->lastinx=key;
       _mi_read_key_record(info, 0L, info->rec_buff);
-      update_auto_increment(info, info->rec_buff);
-      if (info->s->state.auto_increment > save_auto_value)
+      auto_increment= retrieve_auto_increment(info, info->rec_buff);
+      if (auto_increment > info->s->state.auto_increment)
       {
-	mi_check_print_warning(param,
-			       "Auto-increment value: %s is smaller than max used value: %s",
-			       llstr(save_auto_value,buff2),
-			       llstr(info->s->state.auto_increment, buff));
+        mi_check_print_warning(param, "Auto-increment value: %s is smaller "
+                               "than max used value: %s",
+                               llstr(info->s->state.auto_increment,buff2),
+                               llstr(auto_increment, buff));
       }
       if (param->testflag & T_AUTO_INC)
       {
-	set_if_bigger(info->s->state.auto_increment,
-		      param->auto_increment_value);
+        set_if_bigger(info->s->state.auto_increment,
+                      auto_increment);
+        set_if_bigger(info->s->state.auto_increment,
+                      param->auto_increment_value);
       }
-      else
-	info->s->state.auto_increment=save_auto_value;
 
       /* Check that there isn't a row with auto_increment = 0 in the table */
       mi_extra(info,HA_EXTRA_KEYREAD,0);
@@ -481,8 +480,8 @@ int chk_key(MI_CHECK *param, register MI_INFO *info)
       {
 	/* Don't count this as a real warning, as myisamchk can't correct it */
 	uint save=param->warning_printed;
-	mi_check_print_warning(param,
-			       "Found row where the auto_increment column has the value 0");
+        mi_check_print_warning(param, "Found row where the auto_increment "
+                               "column has the value 0");
 	param->warning_printed=save;
       }
       mi_extra(info,HA_EXTRA_NO_KEYREAD,0);
@@ -4124,11 +4123,10 @@ void update_auto_increment_key(MI_CHECK *param, MI_INFO *info,
   }
   else
   {
-    ulonglong auto_increment= (repair_only ? info->s->state.auto_increment :
-			       param->auto_increment_value);
-    info->s->state.auto_increment=0;
-    update_auto_increment(info, record);
+    ulonglong auto_increment= retrieve_auto_increment(info, record);
     set_if_bigger(info->s->state.auto_increment,auto_increment);
+    if (!repair_only)
+      set_if_bigger(info->s->state.auto_increment, param->auto_increment_value);
   }
   mi_extra(info,HA_EXTRA_NO_KEYREAD,0);
   my_free((char*) record, MYF(0));

storage/myisam/mi_key.c

@@ -507,22 +507,21 @@ int _mi_read_key_record(MI_INFO *info, my_off_t filepos, byte *buf)
   return(-1);				/* Wrong data to read */
 }
 
-  
+
 /*
-  Update auto_increment info
+  Retrieve auto_increment info
 
   SYNOPSIS
-    update_auto_increment()
+    retrieve_auto_increment()
     info			MyISAM handler
     record			Row to update
 
   IMPLEMENTATION
-    Only replace the auto_increment value if it is higher than the previous
-    one. For signed columns we don't update the auto increment value if it's
+    For signed columns we don't retrieve the auto increment value if it's
     less than zero.
 */
 
-void update_auto_increment(MI_INFO *info,const byte *record)
+ulonglong retrieve_auto_increment(MI_INFO *info,const byte *record)
 {
   ulonglong value= 0;			/* Store unsigned values here */
   longlong s_value= 0;			/* Store signed values here */
@@ -587,6 +586,5 @@ void update_auto_increment(MI_INFO *info,const byte *record)
     and if s_value == 0 then value will contain either s_value or the
     correct value.
   */
-  set_if_bigger(info->s->state.auto_increment,
-                (s_value > 0) ? (ulonglong) s_value : value);
+  return (s_value > 0) ? (ulonglong) s_value : value;
 }

storage/myisam/mi_update.c

@@ -164,7 +164,8 @@ int mi_update(register MI_INFO *info, const byte *oldrec, byte *newrec)
       key_changed|= HA_STATE_CHANGED;		/* Must update index file */
   }
   if (auto_key_changed)
-    update_auto_increment(info,newrec);
+    set_if_bigger(info->s->state.auto_increment,
+                  retrieve_auto_increment(info, newrec));
   if (share->calc_checksum)
     info->state->checksum+=(info->checksum - old_checksum);
 

storage/myisam/mi_write.c

@@ -149,7 +149,8 @@ int mi_write(MI_INFO *info, byte *record)
     info->state->checksum+=info->checksum;
   }
   if (share->base.auto_key)
-    update_auto_increment(info,record);
+    set_if_bigger(info->s->state.auto_increment,
+                  retrieve_auto_increment(info, record));
   info->update= (HA_STATE_CHANGED | HA_STATE_AKTIV | HA_STATE_WRITTEN |
 		 HA_STATE_ROW_CHANGED);
   info->state->records++;

storage/myisam/myisamdef.h

@@ -593,7 +593,7 @@ extern uint _mi_pack_key(MI_INFO *info,uint keynr,uchar *key,uchar *old,
 extern int _mi_read_key_record(MI_INFO *info,my_off_t filepos,byte *buf);
 extern int _mi_read_cache(IO_CACHE *info,byte *buff,my_off_t pos,
 			  uint length,int re_read_if_possibly);
-extern void update_auto_increment(MI_INFO *info,const byte *record);
+extern ulonglong retrieve_auto_increment(MI_INFO *info,const byte *record);
 
 extern byte *mi_alloc_rec_buff(MI_INFO *,ulong, byte**);
 #define mi_get_rec_buff_ptr(info,buf)                              \