Fixed LP bug #1008293.

One of the reported problems manifested itself in the scenario when one
thread tried to to get statistics on a key cache while the second thread
had not finished initialization of the key cache structure yet. 
The problem was resolved by forcing serialization of such operations
on key caches.

To serialize function calls to perform certain operations over a key cache
a new mutex associated with the key cache now is used. It is stored in the
field op_lock of the KEY_CACHE structure. It is locked when the operation
is performed. Some of the serialized key cache operations utilize calls 
for other key cache operations. To avoid recursive locking of op_lock
the new functions that perform the operations of key cache initialization,
destruction and re-partitioning with an additional parameter were introduced.
The parameter says whether the operation over op_lock are to be performed or
are to be omitted. The old functions for the operations of key cache 
initialization, destruction,and  re-partitioning  now just call the
corresponding new functions with the additional parameter set to true
requesting to use op_lock while all other calls of these new function
have this parameter set to false. 

Another problem reported in the bug entry concerned the operation of
assigning an index to a key cache. This operation can be called
while the key cache structures are not initialized yet. In this
case any call of flush_key_blocks() should return without any actions.

No test case is provided with this patch.

1 files changed, 197 insertions, 55 deletions

diff --git a/mysys/mf_keycache.c b/mysys/mf_keycache.c
index e0d495ce8e0..dc2d7fdd127 100644
--- a/mysys/mf_keycache.c
+++ b/mysys/mf_keycache.c
@@ -5863,31 +5863,31 @@ static KEY_CACHE_FUNCS partitioned_key_cache_funcs =
           
 ******************************************************************************/
 
+static
+int repartition_key_cache_internal(KEY_CACHE *keycache,
+                                   uint key_cache_block_size, size_t use_mem,
+                                   uint division_limit, uint age_threshold,
+                                   uint partitions, my_bool use_op_lock);
 
 /*
-  Initialize a key cache
+  Initialize a key cache : internal
 
   SYNOPSIS
-    init_key_cache()
+    init_key_cache_internal()
     keycache           pointer to the key cache to be initialized
     key_cache_block_size    size of blocks to keep cached data
     use_mem             total memory to use for cache buffers/structures 
     division_limit      division limit (may be zero)
     age_threshold       age threshold (may be zero)
     partitions          number of partitions in the key cache
+    use_op_lock        if TRUE use keycache->op_lock, otherwise - ignore it
 
   DESCRIPTION
-    The function creates a control block structure for a key cache and
-    places the pointer to this block in the structure keycache. 
-    If the value of the parameter 'partitions' is 0 then a simple key cache
-    is created. Otherwise a partitioned key cache with the specified number
-    of partitions is created.  
-    The parameter key_cache_block_size specifies the size of the blocks in
-    the key cache to be created. The parameters division_limit and
-    age_threshold determine the initial values of those characteristics of
-    the key cache that are used for midpoint insertion strategy. The parameter
-    use_mem  specifies the total amount of memory to be allocated for the
-    key cache buffers and for all auxiliary structures.       
+    The function performs the actions required from init_key_cache().
+    It has an additional parameter: use_op_lock. When the parameter
+    is TRUE than the function initializes keycache->op_lock if needed,
+    then locks it, and unlocks it before the return. Otherwise the actions
+    with the lock are omitted. 
 
   RETURN VALUE
     total number of blocks in key cache partitions, if successful,
@@ -5896,19 +5896,22 @@ static KEY_CACHE_FUNCS partitioned_key_cache_funcs =
   NOTES
     if keycache->key_cache_inited != 0 we assume that the memory
     for the control block of the key cache has been already allocated.
-
-    It's assumed that no two threads call this function simultaneously
-    referring to the same key cache handle.
 */
 
-int init_key_cache(KEY_CACHE *keycache, uint key_cache_block_size,
-		   size_t use_mem, uint division_limit,
-		   uint age_threshold, uint partitions)
+static
+int init_key_cache_internal(KEY_CACHE *keycache, uint key_cache_block_size,
+		            size_t use_mem, uint division_limit,
+		            uint age_threshold, uint partitions,
+                            my_bool use_op_lock)
 {
   void *keycache_cb;
   int blocks;
   if (keycache->key_cache_inited)
+  {
+    if (use_op_lock)
+      pthread_mutex_lock(&keycache->op_lock);
     keycache_cb= keycache->keycache_cb;
+  }
   else
   {
     if (partitions == 0)
@@ -5929,8 +5932,17 @@ int init_key_cache(KEY_CACHE *keycache, uint key_cache_block_size,
       keycache->key_cache_type= PARTITIONED_KEY_CACHE;
       keycache->interface_funcs= &partitioned_key_cache_funcs;
     }
+    /*
+      Initialize op_lock if it's not initialized before. 
+      The mutex may have been initialized before if we are being called
+      from repartition_key_cache_internal().
+    */
+    if (use_op_lock)
+      pthread_mutex_init(&keycache->op_lock, MY_MUTEX_INIT_FAST);      
     keycache->keycache_cb= keycache_cb;
     keycache->key_cache_inited= 1;
+    if (use_op_lock)
+      pthread_mutex_lock(&keycache->op_lock);
   }
 
   if (partitions != 0)
@@ -5951,11 +5963,58 @@ int init_key_cache(KEY_CACHE *keycache, uint key_cache_block_size,
     ((SIMPLE_KEY_CACHE_CB *) keycache_cb)->key_cache_mem_size;
   if (blocks > 0)
     keycache->can_be_used= 1;
+  if (use_op_lock)
+    pthread_mutex_unlock(&keycache->op_lock);
   return blocks;
 }
 
 
 /*
+  Initialize a key cache
+
+  SYNOPSIS
+    init_key_cache()
+    keycache           pointer to the key cache to be initialized
+    key_cache_block_size    size of blocks to keep cached data
+    use_mem             total memory to use for cache buffers/structures 
+    division_limit      division limit (may be zero)
+    age_threshold       age threshold (may be zero)
+    partitions          number of partitions in the key cache
+
+  DESCRIPTION
+    The function creates a control block structure for a key cache and
+    places the pointer to this block in the structure keycache. 
+    If the value of the parameter 'partitions' is 0 then a simple key cache
+    is created. Otherwise a partitioned key cache with the specified number
+    of partitions is created.  
+    The parameter key_cache_block_size specifies the size of the blocks in
+    the key cache to be created. The parameters division_limit and
+    age_threshold determine the initial values of those characteristics of
+    the key cache that are used for midpoint insertion strategy. The parameter
+    use_mem  specifies the total amount of memory to be allocated for the
+    key cache buffers and for all auxiliary structures.  
+    The function calls init_key_cache_internal() to perform all these actions
+    with the last parameter set to TRUE.     
+
+  RETURN VALUE
+    total number of blocks in key cache partitions, if successful,
+    <= 0 - otherwise.
+
+  NOTES
+    It's assumed that no two threads call this function simultaneously
+    referring to the same key cache handle.
+*/
+
+int init_key_cache(KEY_CACHE *keycache, uint key_cache_block_size,
+		   size_t use_mem, uint division_limit,
+		   uint age_threshold, uint partitions)
+{
+  return init_key_cache_internal(keycache,  key_cache_block_size, use_mem,
+				 division_limit, age_threshold, partitions, 1);
+}
+
+
+/*
   Resize a key cache
 
   SYNOPSIS
@@ -5995,11 +6054,13 @@ int resize_key_cache(KEY_CACHE *keycache, uint key_cache_block_size,
   int blocks= -1;
   if (keycache->key_cache_inited)
   {
+    pthread_mutex_lock(&keycache->op_lock);
     if ((uint) keycache->param_partitions != keycache->partitions && use_mem)
-      blocks= repartition_key_cache(keycache,
-                                    key_cache_block_size, use_mem,
-                                    division_limit, age_threshold, 
-                                    (uint) keycache->param_partitions);
+      blocks= repartition_key_cache_internal(keycache,
+                                             key_cache_block_size, use_mem,
+                                             division_limit, age_threshold, 
+                                             (uint) keycache->param_partitions,
+                                             0);
     else
     {
       blocks= keycache->interface_funcs->resize(keycache->keycache_cb,
@@ -6018,6 +6079,7 @@ int resize_key_cache(KEY_CACHE *keycache, uint key_cache_block_size,
     ((SIMPLE_KEY_CACHE_CB *)(keycache->keycache_cb))->key_cache_mem_size;
 
     keycache->can_be_used= (blocks >= 0);
+    pthread_mutex_unlock(&keycache->op_lock);
   } 
   return blocks;
 }
@@ -6051,33 +6113,37 @@ void change_key_cache_param(KEY_CACHE *keycache, uint division_limit,
 {
   if (keycache->key_cache_inited)
   {
-    
+    pthread_mutex_lock(&keycache->op_lock);    
     keycache->interface_funcs->change_param(keycache->keycache_cb,
                                             division_limit,
-                                            age_threshold);
+                                            age_threshold);    
+    pthread_mutex_unlock(&keycache->op_lock);
   }
 }
 
 
 /*
-  Destroy a key cache 
+  Destroy a key cache : internal
 
   SYNOPSIS
-    end_key_cache()
+    end_key_cache_internal()
     keycache            pointer to the key cache to be destroyed
     cleanup             <=> complete free 
+    use_op_lock         if TRUE use keycache->op_lock, otherwise - ignore it
 
   DESCRIPTION
-    The function frees the memory allocated for the cache blocks and
-    auxiliary structures used by the key cache keycache. If the value
-    of the parameter cleanup is TRUE then all resources used by the key
-    cache are to be freed.
+    The function performs the actions required from end_key_cache().
+    It has an additional parameter: use_op_lock. When the parameter
+    is TRUE than the function destroys keycache->op_lock if cleanup is true.
+    Otherwise the action with the lock is omitted. 
 
   RETURN VALUE
     none
 */
 
-void end_key_cache(KEY_CACHE *keycache, my_bool cleanup)
+static
+void end_key_cache_internal(KEY_CACHE *keycache, my_bool cleanup,
+                            my_bool use_op_lock)
 {
   if (keycache->key_cache_inited)
   {
@@ -6089,6 +6155,12 @@ void end_key_cache(KEY_CACHE *keycache, my_bool cleanup)
         my_free((uchar *) keycache->keycache_cb, MYF(0));
         keycache->keycache_cb= 0;
       }
+      /*
+        We do not destroy op_lock if we are going to reuse the same key cache.
+        This happens if we are called from  repartition_key_cache_internal().
+      */
+      if (use_op_lock)
+        pthread_mutex_destroy(&keycache->op_lock);
       keycache->key_cache_inited= 0;
     }
     keycache->can_be_used= 0;
@@ -6097,6 +6169,32 @@ void end_key_cache(KEY_CACHE *keycache, my_bool cleanup)
 
 
 /*
+  Destroy a key cache 
+
+  SYNOPSIS
+    end_key_cache()
+    keycache            pointer to the key cache to be destroyed
+    cleanup             <=> complete free 
+
+  DESCRIPTION
+    The function frees the memory allocated for the cache blocks and
+    auxiliary structures used by the key cache keycache. If the value
+    of the parameter cleanup is TRUE then all resources used by the key
+    cache are to be freed.
+    The function calls end_key_cache_internal() to perform all these actions
+    with the last parameter set to TRUE.     
+
+  RETURN VALUE
+    none
+*/
+
+void end_key_cache(KEY_CACHE *keycache, my_bool cleanup)
+{
+  end_key_cache_internal(keycache, cleanup, 1);
+}
+
+
+/*
   Read a block of data from a key cache into a buffer
 
   SYNOPSIS
@@ -6140,7 +6238,7 @@ uchar *key_cache_read(KEY_CACHE *keycache,
                       uchar *buff, uint length,
 		      uint block_length, int return_buffer)
 {
-  if (keycache->key_cache_inited && keycache->can_be_used)
+  if (keycache->can_be_used)
     return keycache->interface_funcs->read(keycache->keycache_cb,
                                            file, filepos, level,
                                            buff, length,
@@ -6192,7 +6290,7 @@ int key_cache_insert(KEY_CACHE *keycache,
                      File file, my_off_t filepos, int level,
                      uchar *buff, uint length)
 {
-  if (keycache->key_cache_inited && keycache->can_be_used)
+  if (keycache->can_be_used)
     return keycache->interface_funcs->insert(keycache->keycache_cb,
                                              file, filepos, level,
                                              buff, length);
@@ -6247,7 +6345,7 @@ int key_cache_write(KEY_CACHE *keycache,
                     uchar *buff, uint length,
 		    uint block_length, int force_write)
 {
-  if (keycache->key_cache_inited && keycache->can_be_used)
+  if (keycache->can_be_used)
     return keycache->interface_funcs->write(keycache->keycache_cb,
                                             file, file_extra,
                                             filepos, level,
@@ -6299,7 +6397,7 @@ int flush_key_blocks(KEY_CACHE *keycache,
                      int file, void *file_extra,
                      enum flush_type type)
 {
-  if (keycache->key_cache_inited)
+  if (keycache->can_be_used)
     return keycache->interface_funcs->flush(keycache->keycache_cb,
                                             file, file_extra, type);
   return 0;  
@@ -6330,13 +6428,15 @@ int flush_key_blocks(KEY_CACHE *keycache,
 int reset_key_cache_counters(const char *name __attribute__((unused)),
                              KEY_CACHE *keycache)
 {
+  int rc= 0;
   if (keycache->key_cache_inited)
   {
-    
-    return keycache->interface_funcs->reset_counters(name,
-                                                     keycache->keycache_cb);
+    pthread_mutex_lock(&keycache->op_lock);
+    rc= keycache->interface_funcs->reset_counters(name,
+                                                  keycache->keycache_cb);
+    pthread_mutex_unlock(&keycache->op_lock);
   }
-  return 0;
+  return rc;
 }
 
 
@@ -6366,11 +6466,63 @@ void get_key_cache_statistics(KEY_CACHE *keycache, uint partition_no,
 {
   if (keycache->key_cache_inited)
   {    
+    pthread_mutex_lock(&keycache->op_lock);
     keycache->interface_funcs->get_stats(keycache->keycache_cb,
                                          partition_no, key_cache_stats);
+    pthread_mutex_unlock(&keycache->op_lock);
   }
 }
 
+
+/*
+  Repartition a key cache : internal
+
+  SYNOPSIS
+    repartition_key_cache_internal()
+    keycache           pointer to the key cache to be repartitioned
+    key_cache_block_size    size of blocks to keep cached data
+    use_mem             total memory to use for the new key cache
+    division_limit      new division limit (if not zero)
+    age_threshold       new age threshold (if not zero)
+    partitions          new number of partitions in the key cache 
+    use_op_lock         if TRUE use keycache->op_lock, otherwise - ignore it
+
+  DESCRIPTION
+    The function performs the actions required from repartition_key_cache().
+    It has an additional parameter: use_op_lock. When the parameter
+    is TRUE then the function locks keycache->op_lock at start and
+    unlocks it before the return. Otherwise the actions with the lock
+    are omitted. 
+
+  RETURN VALUE
+    number of blocks in the key cache, if successful,
+    0 - otherwise.
+*/
+
+static
+int repartition_key_cache_internal(KEY_CACHE *keycache,
+                                   uint key_cache_block_size, size_t use_mem,
+                                   uint division_limit, uint age_threshold,
+                                   uint partitions, my_bool use_op_lock)
+{
+  uint blocks= -1;
+  if (keycache->key_cache_inited)
+  {
+    if (use_op_lock)
+      pthread_mutex_lock(&keycache->op_lock);
+    keycache->interface_funcs->resize(keycache->keycache_cb,
+                                      key_cache_block_size, 0,
+                                      division_limit, age_threshold);
+    end_key_cache_internal(keycache, 1, 0);
+    blocks= init_key_cache_internal(keycache, key_cache_block_size, use_mem,
+                                    division_limit, age_threshold, partitions,
+                                    0);
+    if (use_op_lock)
+      pthread_mutex_unlock(&keycache->op_lock);
+  } 
+  return blocks;
+}
+
 /*
   Repartition a key cache
 
@@ -6394,16 +6546,14 @@ void get_key_cache_statistics(KEY_CACHE *keycache, uint partition_no,
     that are used for midpoint insertion strategy. The parameter use_mem
     specifies the total amount of  memory to be allocated for the new key
     cache buffers and for all auxiliary structures.
+    The function calls repartition_key_cache_internal() to perform all these
+    actions with the last parameter set to TRUE.     
 
   RETURN VALUE
     number of blocks in the key cache, if successful,
     0 - otherwise.
 
   NOTES
-    The function does not block the calls and executions of other functions
-    from the key cache interface. However it assumes that the calls of 
-    resize_key_cache itself are serialized.
-
     Currently the function is called when the value of the variable
     key_cache_partitions is being reset for the key cache keycache.
 */
@@ -6412,16 +6562,8 @@ int repartition_key_cache(KEY_CACHE *keycache, uint key_cache_block_size,
 		          size_t use_mem, uint division_limit,
                           uint age_threshold, uint partitions)
 {
-  uint blocks= -1;
-  if (keycache->key_cache_inited)
-  {
-    keycache->interface_funcs->resize(keycache->keycache_cb,
-                                      key_cache_block_size, 0,
-                                      division_limit, age_threshold);
-    end_key_cache(keycache, 1);
-    blocks= init_key_cache(keycache, key_cache_block_size, use_mem,
-                           division_limit, age_threshold, partitions);
-  } 
-  return blocks;
+  return repartition_key_cache_internal(keycache, key_cache_block_size, use_mem,
+			                division_limit, age_threshold,
+                                        partitions, 1);
 }