A fix for Bug#7209 "Client error with "Access Denied" on updates

when high concurrency": remove HASH::current_record and make it
an external search parameter, so that it can not be the cause of a 
race condition under high concurrent load.
The bug was in a race condition in table_hash_search,
when column_priv_hash.current_record was overwritten simultaneously
by multiple threads, causing the search for a suitable grant record
to fail.
No test case as the bug is repeatable only under concurrent load.


include/hash.h:
  - remove current_record from HASH, instead modify hash_first,
  hash_next to accept HASH_SEARCH_STATE as an IN/OUT parameter
mysys/hash.c:
  - remove HASH::current_record
  - change declarations of functions that use HASH in read-only mode
    to accept const HASH * instead of HASH *. 
  - implement hash_search; move the old implementation of hash_search
  to hash_first
mysys/testhash.c:
  - adjust the test case to changed function declarations
sql/lock.cc:
  - adjust to changed declarations of hash_search, hash_next
sql/sql_acl.cc:
  - adjust to changed declarations of hash_search, hash_next
sql/sql_base.cc:
  - adjust to changed declarations of hash_search, hash_nex
sql/sql_cache.cc:
  - adjust to a changed declaration of hash_replace

include/hash.h

@@ -33,7 +33,7 @@ typedef void (*hash_free_key)(void *);
 
 typedef struct st_hash {
   uint key_offset,key_length;		/* Length of key if const length */
-  uint records,blength,current_record;
+  uint records, blength;
   uint flags;
   DYNAMIC_ARRAY array;				/* Place for hash_keys */
   hash_get_key get_key;
@@ -41,6 +41,9 @@ typedef struct st_hash {
   CHARSET_INFO *charset;
 } HASH;
 
+/* A search iterator state */
+typedef uint HASH_SEARCH_STATE;
+
 #define hash_init(A,B,C,D,E,F,G,H) _hash_init(A,B,C,D,E,F,G, H CALLER_INFO)
 my_bool _hash_init(HASH *hash, CHARSET_INFO *charset,
 		   uint default_array_elements, uint key_offset,
@@ -49,12 +52,15 @@ my_bool _hash_init(HASH *hash, CHARSET_INFO *charset,
 void hash_free(HASH *tree);
 void my_hash_reset(HASH *hash);
 byte *hash_element(HASH *hash,uint idx);
-gptr hash_search(HASH *info,const byte *key,uint length);
-gptr hash_next(HASH *info,const byte *key,uint length);
+gptr hash_search(const HASH *info, const byte *key, uint length);
+gptr hash_first(const HASH *info, const byte *key, uint length,
+                HASH_SEARCH_STATE *state);
+gptr hash_next(const HASH *info, const byte *key, uint length,
+               HASH_SEARCH_STATE *state);
 my_bool my_hash_insert(HASH *info,const byte *data);
 my_bool hash_delete(HASH *hash,byte *record);
 my_bool hash_update(HASH *hash,byte *record,byte *old_key,uint old_key_length);
-void hash_replace(HASH *hash, uint idx, byte *new_row);
+void hash_replace(HASH *hash, HASH_SEARCH_STATE *state, byte *new_row);
 my_bool hash_check(HASH *hash);			/* Only in debug library */
 
 #define hash_clear(H) bzero((char*) (H),sizeof(*(H)))

mysys/hash.c

@@ -36,9 +36,10 @@ typedef struct st_hash_info {
 
 static uint hash_mask(uint hashnr,uint buffmax,uint maxlength);
 static void movelink(HASH_LINK *array,uint pos,uint next_link,uint newlink);
-static int hashcmp(HASH *hash,HASH_LINK *pos,const byte *key,uint length);
+static int hashcmp(const HASH *hash, HASH_LINK *pos, const byte *key,
+                   uint length);
 
-static uint calc_hash(HASH *hash,const byte *key,uint length)
+static uint calc_hash(const HASH *hash, const byte *key, uint length)
 {
   ulong nr1=1, nr2=4;
   hash->charset->coll->hash_sort(hash->charset,(uchar*) key,length,&nr1,&nr2);
@@ -63,7 +64,6 @@ _hash_init(HASH *hash,CHARSET_INFO *charset,
   hash->key_offset=key_offset;
   hash->key_length=key_length;
   hash->blength=1;
-  hash->current_record= NO_RECORD;		/* For the future */
   hash->get_key=get_key;
   hash->free=free_element;
   hash->flags=flags;
@@ -135,7 +135,6 @@ void my_hash_reset(HASH *hash)
   reset_dynamic(&hash->array);
   /* Set row pointers so that the hash can be reused at once */
   hash->blength= 1;
-  hash->current_record= NO_RECORD;
   DBUG_VOID_RETURN;
 }
 
@@ -147,7 +146,8 @@ void my_hash_reset(HASH *hash)
 */
 
 static inline char*
-hash_key(HASH *hash,const byte *record,uint *length,my_bool first)
+hash_key(const HASH *hash, const byte *record, uint *length,
+         my_bool first)
 {
   if (hash->get_key)
     return (*hash->get_key)(record,length,first);
@@ -163,8 +163,8 @@ static uint hash_mask(uint hashnr,uint buffmax,uint maxlength)
   return (hashnr & ((buffmax >> 1) -1));
 }
 
-static uint hash_rec_mask(HASH *hash,HASH_LINK *pos,uint buffmax,
-			  uint maxlength)
+static uint hash_rec_mask(const HASH *hash, HASH_LINK *pos,
+                          uint buffmax, uint maxlength)
 {
   uint length;
   byte *key= (byte*) hash_key(hash,pos->data,&length,0);
@@ -186,14 +186,25 @@ unsigned int rec_hashnr(HASH *hash,const byte *record)
 }
 
 
-	/* Search after a record based on a key */
-	/* Sets info->current_ptr to found record */
+gptr hash_search(const HASH *hash, const byte *key, uint length)
+{
+  HASH_SEARCH_STATE state;
+  return hash_first(hash, key, length, &state);
+}
+
+/*
+  Search after a record based on a key
+
+  NOTE
+   Assigns the number of the found record to HASH_SEARCH_STATE state
+*/
 
-gptr hash_search(HASH *hash,const byte *key,uint length)
+gptr hash_first(const HASH *hash, const byte *key, uint length,
+                HASH_SEARCH_STATE *current_record)
 {
   HASH_LINK *pos;
   uint flag,idx;
-  DBUG_ENTER("hash_search");
+  DBUG_ENTER("hash_first");
 
   flag=1;
   if (hash->records)
@@ -206,7 +217,7 @@ gptr hash_search(HASH *hash,const byte *key,uint length)
       if (!hashcmp(hash,pos,key,length))
       {
 	DBUG_PRINT("exit",("found key at %d",idx));
-	hash->current_record= idx;
+	*current_record= idx;
 	DBUG_RETURN (pos->data);
       }
       if (flag)
@@ -218,31 +229,32 @@ gptr hash_search(HASH *hash,const byte *key,uint length)
     }
     while ((idx=pos->next) != NO_RECORD);
   }
-  hash->current_record= NO_RECORD;
+  *current_record= NO_RECORD;
   DBUG_RETURN(0);
 }
 
 	/* Get next record with identical key */
 	/* Can only be called if previous calls was hash_search */
 
-gptr hash_next(HASH *hash,const byte *key,uint length)
+gptr hash_next(const HASH *hash, const byte *key, uint length,
+               HASH_SEARCH_STATE *current_record)
 {
   HASH_LINK *pos;
   uint idx;
 
-  if (hash->current_record != NO_RECORD)
+  if (*current_record != NO_RECORD)
   {
     HASH_LINK *data=dynamic_element(&hash->array,0,HASH_LINK*);
-    for (idx=data[hash->current_record].next; idx != NO_RECORD ; idx=pos->next)
+    for (idx=data[*current_record].next; idx != NO_RECORD ; idx=pos->next)
     {
       pos=data+idx;
       if (!hashcmp(hash,pos,key,length))
       {
-	hash->current_record= idx;
+	*current_record= idx;
 	return pos->data;
       }
     }
-    hash->current_record=NO_RECORD;
+    *current_record= NO_RECORD;
   }
   return 0;
 }
@@ -282,7 +294,8 @@ static void movelink(HASH_LINK *array,uint find,uint next_link,uint newlink)
     > 0  key of record >  key
  */
 
-static int hashcmp(HASH *hash,HASH_LINK *pos,const byte *key,uint length)
+static int hashcmp(const HASH *hash, HASH_LINK *pos, const byte *key,
+                   uint length)
 {
   uint rec_keylength;
   byte *rec_key= (byte*) hash_key(hash,pos->data,&rec_keylength,1);
@@ -308,7 +321,6 @@ my_bool my_hash_insert(HASH *info,const byte *record)
   if (!(empty=(HASH_LINK*) alloc_dynamic(&info->array)))
     return(TRUE);				/* No more memory */
 
-  info->current_record= NO_RECORD;
   data=dynamic_element(&info->array,0,HASH_LINK*);
   halfbuff= info->blength >> 1;
 
@@ -451,7 +463,6 @@ my_bool hash_delete(HASH *hash,byte *record)
   }
 
   if ( --(hash->records) < hash->blength >> 1) hash->blength>>=1;
-  hash->current_record= NO_RECORD;
   lastpos=data+hash->records;
 
   /* Remove link to record */
@@ -544,7 +555,6 @@ my_bool hash_update(HASH *hash,byte *record,byte *old_key,uint old_key_length)
     if ((idx=pos->next) == NO_RECORD)
       DBUG_RETURN(1);			/* Not found in links */
   }
-  hash->current_record= NO_RECORD;
   org_link= *pos;
   empty=idx;
 
@@ -594,10 +604,10 @@ byte *hash_element(HASH *hash,uint idx)
   isn't changed
 */
 
-void hash_replace(HASH *hash, uint idx, byte *new_row)
+void hash_replace(HASH *hash, HASH_SEARCH_STATE *current_record, byte *new_row)
 {
-  if (idx != NO_RECORD)				/* Safety */
-    dynamic_element(&hash->array,idx,HASH_LINK*)->data=new_row;
+  if (*current_record != NO_RECORD)            /* Safety */
+    dynamic_element(&hash->array, *current_record, HASH_LINK*)->data= new_row;
 }
 
 

mysys/testhash.c

@@ -74,7 +74,7 @@ static int do_test()
   bzero((char*) key1,sizeof(key1[0])*1000);
 
   printf("- Creating hash\n");
-  if (hash_init(&hash,recant/2,0,6,0,free_record,0))
+  if (hash_init(&hash, default_charset_info, recant/2, 0, 6, 0, free_record, 0))
     goto err;
   printf("- Writing records:\n");
 
@@ -172,15 +172,16 @@ static int do_test()
       break;
   if (key1[j] > 1)
   {
+    HASH_SEARCH_STATE state;
     printf("- Testing identical read\n");
     sprintf(key,"%6d",j);
     pos=1;
-    if (!(recpos=hash_search(&hash,key,0)))
+    if (!(recpos= hash_first(&hash, key, 0, &state)))
     {
       printf("can't find key1: \"%s\"\n",key);
       goto err;
     }
-    while (hash_next(&hash,key,0) && pos < (ulong) (key1[j]+10))
+    while (hash_next(&hash, key, 0, &state) && pos < (ulong) (key1[j]+10))
       pos++;
     if (pos != (ulong) key1[j])
     {
@@ -189,7 +190,7 @@ static int do_test()
     }
   }
   printf("- Creating output heap-file 2\n");
-  if (hash_init(&hash2,hash.records,0,0,hash2_key,free_record,0))
+  if (hash_init(&hash2, default_charset_info, hash.records, 0, 0, hash2_key, free_record,0))
     goto err;
 
   printf("- Copying and removing records\n");

sql/lock.cc

@@ -641,6 +641,7 @@ int lock_table_name(THD *thd, TABLE_LIST *table_list)
   char  key[MAX_DBKEY_LENGTH];
   char *db= table_list->db;
   uint  key_length;
+  HASH_SEARCH_STATE state;
   DBUG_ENTER("lock_table_name");
   DBUG_PRINT("enter",("db: %s  name: %s", db, table_list->real_name));
 
@@ -651,9 +652,9 @@ int lock_table_name(THD *thd, TABLE_LIST *table_list)
 
 
   /* Only insert the table if we haven't insert it already */
-  for (table=(TABLE*) hash_search(&open_cache,(byte*) key,key_length) ;
+  for (table=(TABLE*) hash_first(&open_cache, (byte*)key, key_length, &state);
        table ;
-       table = (TABLE*) hash_next(&open_cache,(byte*) key,key_length))
+       table = (TABLE*) hash_next(&open_cache, (byte*)key, key_length, &state))
     if (table->in_use == thd)
       DBUG_RETURN(0);
 

sql/sql_acl.cc

@@ -1988,14 +1988,15 @@ static GRANT_TABLE *table_hash_search(const char *host,const char* ip,
   char helping [NAME_LEN*2+USERNAME_LENGTH+3];
   uint len;
   GRANT_TABLE *grant_table,*found=0;
+  HASH_SEARCH_STATE state;
 
   len  = (uint) (strmov(strmov(strmov(helping,user)+1,db)+1,tname)-helping)+ 1;
-  for (grant_table=(GRANT_TABLE*) hash_search(&column_priv_hash,
+  for (grant_table=(GRANT_TABLE*) hash_first(&column_priv_hash,
                                              (byte*) helping,
-					      len) ;
+                                             len, &state) ;
        grant_table ;
        grant_table= (GRANT_TABLE*) hash_next(&column_priv_hash,(byte*) helping,
-					     len))
+                                             len, &state))
   {
     if (exact)
     {

sql/sql_base.cc

@@ -799,6 +799,7 @@ TABLE *open_table(THD *thd,const char *db,const char *table_name,
   reg1	TABLE *table;
   char	key[MAX_DBKEY_LENGTH];
   uint	key_length;
+  HASH_SEARCH_STATE state;
   DBUG_ENTER("open_table");
 
   /* find a unused table in the open table cache */
@@ -863,9 +864,11 @@ TABLE *open_table(THD *thd,const char *db,const char *table_name,
   /* close handler tables which are marked for flush */
   mysql_ha_flush(thd, (TABLE_LIST*) NULL, MYSQL_HA_REOPEN_ON_USAGE, TRUE);
 
-  for (table=(TABLE*) hash_search(&open_cache,(byte*) key,key_length) ;
+  for (table= (TABLE*) hash_first(&open_cache, (byte*) key, key_length,
+                                  &state);
        table && table->in_use ;
-       table = (TABLE*) hash_next(&open_cache,(byte*) key,key_length))
+       table= (TABLE*) hash_next(&open_cache, (byte*) key, key_length,
+                                 &state))
   {
     if (table->version != refresh_version)
     {
@@ -1236,12 +1239,14 @@ bool table_is_used(TABLE *table, bool wait_for_name_lock)
 {
   do
   {
+    HASH_SEARCH_STATE state;
     char *key= table->table_cache_key;
     uint key_length=table->key_length;
-    for (TABLE *search=(TABLE*) hash_search(&open_cache,
-					    (byte*) key,key_length) ;
+    for (TABLE *search= (TABLE*) hash_first(&open_cache, (byte*) key,
+                                            key_length, &state);
 	 search ;
-	 search = (TABLE*) hash_next(&open_cache,(byte*) key,key_length))
+         search= (TABLE*) hash_next(&open_cache, (byte*) key,
+                                    key_length, &state))
     {
       if (search->locked_by_flush ||
 	  search->locked_by_name && wait_for_name_lock ||
@@ -2958,11 +2963,14 @@ bool remove_table_from_cache(THD *thd, const char *db, const char *table_name,
   key_length=(uint) (strmov(strmov(key,db)+1,table_name)-key)+1;
   for (;;)
   {
+    HASH_SEARCH_STATE state;
     result= signalled= 0;
 
-    for (table=(TABLE*) hash_search(&open_cache,(byte*) key,key_length) ;
+    for (table= (TABLE*) hash_first(&open_cache, (byte*) key, key_length,
+                                    &state);
          table;
-         table = (TABLE*) hash_next(&open_cache,(byte*) key,key_length))
+         table= (TABLE*) hash_next(&open_cache, (byte*) key, key_length,
+                                   &state))
     {
       THD *in_use;
       table->version=0L;		/* Free when thread is ready */

sql/sql_cache.cc

@@ -2873,6 +2873,7 @@ my_bool Query_cache::move_by_type(byte **border,
   }
   case Query_cache_block::TABLE:
   {
+    HASH_SEARCH_STATE record_idx;
     DBUG_PRINT("qcache", ("block 0x%lx TABLE", (ulong) block));
     if (*border == 0)
       break;
@@ -2890,7 +2891,7 @@ my_bool Query_cache::move_by_type(byte **border,
     byte *key;
     uint key_length;
     key=query_cache_table_get_key((byte*) block, &key_length, 0);
-    hash_search(&tables, (byte*) key, key_length);
+    hash_first(&tables, (byte*) key, key_length, &record_idx);
 
     block->destroy();
     new_block->init(len);
@@ -2924,7 +2925,7 @@ my_bool Query_cache::move_by_type(byte **border,
     /* Fix pointer to table name */
     new_block->table()->table(new_block->table()->db() + tablename_offset);
     /* Fix hash to point at moved block */
-    hash_replace(&tables, tables.current_record, (byte*) new_block);
+    hash_replace(&tables, &record_idx, (byte*) new_block);
 
     DBUG_PRINT("qcache", ("moved %lu bytes to 0x%lx, new gap at 0x%lx",
 			len, (ulong) new_block, (ulong) *border));
@@ -2932,6 +2933,7 @@ my_bool Query_cache::move_by_type(byte **border,
   }
   case Query_cache_block::QUERY:
   {
+    HASH_SEARCH_STATE record_idx;
     DBUG_PRINT("qcache", ("block 0x%lx QUERY", (ulong) block));
     if (*border == 0)
       break;
@@ -2949,7 +2951,7 @@ my_bool Query_cache::move_by_type(byte **border,
     byte *key;
     uint key_length;
     key=query_cache_query_get_key((byte*) block, &key_length, 0);
-    hash_search(&queries, (byte*) key, key_length);
+    hash_first(&queries, (byte*) key, key_length, &record_idx);
     // Move table of used tables 
     memmove((char*) new_block->table(0), (char*) block->table(0),
 	   ALIGN_SIZE(n_tables*sizeof(Query_cache_block_table)));
@@ -3017,7 +3019,7 @@ my_bool Query_cache::move_by_type(byte **border,
       net->query_cache_query= (gptr) new_block;
     }
     /* Fix hash to point at moved block */
-    hash_replace(&queries, queries.current_record, (byte*) new_block);
+    hash_replace(&queries, &record_idx, (byte*) new_block);
     DBUG_PRINT("qcache", ("moved %lu bytes to 0x%lx, new gap at 0x%lx",
 			len, (ulong) new_block, (ulong) *border));
     break;