Fixed compiler & valgrind warnings from my previous push.

Fixed a bug in Aria when two threads was inserting into the same table and row page and one thread did an abort becasue of duplicate key.



mysys/thr_lock.c:
  Fixed valgrind warning
sql/sql_base.cc:
  Remove not used variable
storage/maria/ma_bitmap.c:
  Added ma_bitmap_lock() & ma_bitmap_unlock() to protect against two threads using the bitmap at the same time.
  More DBUG_PRINT()
storage/maria/ma_blockrec.c:
  Fixed a bug in Aria when two threads was inserting into the same table and row page and one thread did an abort becasue of duplicate key.
  Fix was that we block other threads to modify the bitmap while we are removing the row with a duplicate key.
storage/maria/ma_blockrec.h:
  Added ma_bitmap_lock() & ma_bitmap_unlock() to protect against two threads using the bitmap at the same time.
storage/maria/maria_def.h:
  Changed flush_all_requested to be a counter.
storage/myisam/mi_locking.c:
  Fixed compiler error on windows (typo).

4 files changed, 84 insertions, 9 deletions

diff --git a/storage/maria/ma_bitmap.c b/storage/maria/ma_bitmap.c
index c96e0bd7a86..c0763b0612d 100644
--- a/storage/maria/ma_bitmap.c
+++ b/storage/maria/ma_bitmap.c
@@ -227,7 +227,8 @@ my_bool _ma_bitmap_init(MARIA_SHARE *share, File file)
     The +1 is to add the bitmap page, as this doesn't have to be covered
   */
   bitmap->pages_covered= aligned_bit_blocks * 16 + 1;
-  bitmap->flush_all_requested= bitmap->non_flushable= 0;
+  bitmap->flush_all_requested= 0;
+  bitmap->non_flushable= 0;
 
   /* Update size for bits */
   /* TODO; Make this dependent of the row size */
@@ -311,8 +312,8 @@ my_bool _ma_bitmap_flush(MARIA_SHARE *share)
     pthread_mutex_lock(&share->bitmap.bitmap_lock);
     if (share->bitmap.changed)
     {
-      share->bitmap.changed= 0;
       res= write_changed_bitmap(share, &share->bitmap);
+      share->bitmap.changed= 0;
     }
     pthread_mutex_unlock(&share->bitmap.bitmap_lock);
   }
@@ -355,7 +356,7 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
   pthread_mutex_lock(&bitmap->bitmap_lock);
   if (bitmap->changed || bitmap->changed_not_flushed)
   {
-    bitmap->flush_all_requested= TRUE;
+    bitmap->flush_all_requested++;
 #ifndef WRONG_BITMAP_FLUSH
     while (bitmap->non_flushable > 0)
     {
@@ -363,6 +364,7 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
       pthread_cond_wait(&bitmap->bitmap_cond, &bitmap->bitmap_lock);
     }
 #endif
+    DBUG_ASSERT(bitmap->flush_all_requested == 1);
     /*
       Bitmap is in a flushable state: its contents in memory are reflected by
       log records (complete REDO-UNDO groups) and all bitmap pages are
@@ -391,7 +393,7 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
         PCFLUSH_PINNED_AND_ERROR)
       res= TRUE;
     bitmap->changed_not_flushed= FALSE;
-    bitmap->flush_all_requested= FALSE;
+    bitmap->flush_all_requested--;
     /*
       Some well-behaved threads may be waiting for flush_all_requested to
       become false, wake them up.
@@ -405,6 +407,70 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
 
 
 /**
+   @brief Lock bitmap from being used by another thread
+
+   @fn _ma_bitmap_lock()
+   @param  share               Table's share
+
+   @notes
+   This is a temporary solution for allowing someone to delete an inserted
+   duplicate-key row while someone else is doing concurrent inserts.
+   This is ok for now as duplicate key errors are not that common.
+
+   In the future we will add locks for row-pages to ensure two threads doesn't
+   work at the same time on the same page.
+*/
+
+void _ma_bitmap_lock(MARIA_SHARE *share)
+{
+  MARIA_FILE_BITMAP *bitmap= &share->bitmap;
+  DBUG_ENTER("_ma_bitmap_lock");
+
+  if (!share->now_transactional)
+    DBUG_VOID_RETURN;
+
+  pthread_mutex_lock(&bitmap->bitmap_lock);
+  bitmap->flush_all_requested++;
+  while (bitmap->non_flushable)
+  {
+    DBUG_PRINT("info", ("waiting for bitmap to be flushable"));
+    pthread_cond_wait(&bitmap->bitmap_cond, &bitmap->bitmap_lock);
+  }
+  /*
+    Ensure that _ma_bitmap_flush_all() and _ma_bitmap_lock() are blocked.
+    ma_bitmap_flushable() is blocked thanks to 'flush_all_requested'.
+  */
+  bitmap->non_flushable= 1;
+  pthread_mutex_unlock(&bitmap->bitmap_lock);
+  DBUG_VOID_RETURN;
+}
+  
+/**
+   @brief Unlock bitmap after _ma_bitmap_lock()
+
+   @fn _ma_bitmap_unlock()
+   @param  share               Table's share
+*/
+
+void _ma_bitmap_unlock(MARIA_SHARE *share)
+{
+  MARIA_FILE_BITMAP *bitmap= &share->bitmap;
+  DBUG_ENTER("_ma_bitmap_unlock");
+
+  if (!share->now_transactional)
+    DBUG_VOID_RETURN;
+  DBUG_ASSERT(bitmap->flush_all_requested > 0 && bitmap->non_flushable == 1);
+
+  pthread_mutex_lock(&bitmap->bitmap_lock);
+  bitmap->flush_all_requested--;
+  bitmap->non_flushable= 0;
+  pthread_mutex_unlock(&bitmap->bitmap_lock);
+  pthread_cond_broadcast(&bitmap->bitmap_cond);
+  DBUG_VOID_RETURN;
+}
+
+
+/**
   @brief Unpin all pinned bitmap pages
 
   @param  share            Table's share
@@ -633,11 +699,12 @@ static void _ma_print_bitmap_changes(MARIA_FILE_BITMAP *bitmap)
 {
   uchar *pos, *end, *org_pos;
   ulong page;
+  DBUG_ENTER("_ma_print_bitmap_changes");
 
   end= bitmap->map + bitmap->used_size;
   DBUG_LOCK_FILE;
-  fprintf(DBUG_FILE,"\nBitmap page changes at page %lu\n",
-          (ulong) bitmap->page);
+  fprintf(DBUG_FILE,"\nBitmap page changes at page: %lu  bitmap: 0x%lx\n",
+          (ulong) bitmap->page, (long) bitmap->map);
 
   page= (ulong) bitmap->page+1;
   for (pos= bitmap->map, org_pos= bitmap->map + bitmap->block_size ;
@@ -666,6 +733,7 @@ static void _ma_print_bitmap_changes(MARIA_FILE_BITMAP *bitmap)
   fputc('\n', DBUG_FILE);
   DBUG_UNLOCK_FILE;
   memcpy(bitmap->map + bitmap->block_size, bitmap->map, bitmap->block_size);
+  DBUG_VOID_RETURN;
 }
 
 
@@ -877,6 +945,7 @@ static void fill_block(MARIA_FILE_BITMAP *bitmap,
 {
   uint page, offset, tmp;
   uchar *data;
+  DBUG_ENTER("fill_block");
 
   /* For each 6 bytes we have 6*8/3= 16 patterns */
   page= ((uint) (best_data - bitmap->map)) / 6 * 16 + best_pos;
@@ -902,6 +971,7 @@ static void fill_block(MARIA_FILE_BITMAP *bitmap,
   int2store(data, tmp);
   bitmap->changed= 1;
   DBUG_EXECUTE("bitmap", _ma_print_bitmap_changes(bitmap););
+  DBUG_VOID_RETURN;
 }
 
 
@@ -1514,6 +1584,8 @@ static void use_head(MARIA_HA *info, pgcache_page_no_t page, uint size,
   MARIA_BITMAP_BLOCK *block;
   uchar *data;
   uint offset, tmp, offset_page;
+  DBUG_ENTER("use_head");
+
   DBUG_ASSERT(page % bitmap->pages_covered);
 
   block= dynamic_element(&info->bitmap_blocks, block_position,
@@ -1536,6 +1608,7 @@ static void use_head(MARIA_HA *info, pgcache_page_no_t page, uint size,
   int2store(data, tmp);
   bitmap->changed= 1;
   DBUG_EXECUTE("bitmap", _ma_print_bitmap_changes(bitmap););
+  DBUG_VOID_RETURN;
 }
 
 
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index bc2b1d5187f..0eab7c62796 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -3592,7 +3592,7 @@ my_bool _ma_write_abort_block_record(MARIA_HA *info)
   MARIA_SHARE *share= info->s;
   DBUG_ENTER("_ma_write_abort_block_record");
 
-  _ma_bitmap_flushable(info, 1);
+  _ma_bitmap_lock(share);  /* Lock bitmap from other insert threads */
   if (delete_head_or_tail(info,
                           ma_recordpos_to_page(info->cur_row.lastpos),
                           ma_recordpos_to_dir_entry(info->cur_row.lastpos), 1,
@@ -3630,7 +3630,7 @@ my_bool _ma_write_abort_block_record(MARIA_HA *info)
                       &lsn, (void*) 0))
       res= 1;
   }
-  _ma_bitmap_flushable(info, -1);
+  _ma_bitmap_unlock(share);
   _ma_unpin_all_pages_and_finalize_row(info, lsn);
   DBUG_RETURN(res);
 }
diff --git a/storage/maria/ma_blockrec.h b/storage/maria/ma_blockrec.h
index d18b20aa387..a5858880dd0 100644
--- a/storage/maria/ma_blockrec.h
+++ b/storage/maria/ma_blockrec.h
@@ -217,6 +217,8 @@ uint _ma_bitmap_get_page_bits(MARIA_HA *info, MARIA_FILE_BITMAP *bitmap,
 void _ma_bitmap_delete_all(MARIA_SHARE *share);
 int  _ma_bitmap_create_first(MARIA_SHARE *share);
 void _ma_bitmap_flushable(MARIA_HA *info, int non_flushable_inc);
+void _ma_bitmap_lock(MARIA_SHARE *share);
+void _ma_bitmap_unlock(MARIA_SHARE *share);
 void _ma_bitmap_set_pagecache_callbacks(PAGECACHE_FILE *file,
                                         MARIA_SHARE *share);
 #ifndef DBUG_OFF
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index 6be34aae98c..bdc4d4c5ed2 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -246,7 +246,7 @@ typedef struct st_maria_file_bitmap
   uint used_size;                      /* Size of bitmap head that is not 0 */
   my_bool changed;                     /* 1 if page needs to be written */
   my_bool changed_not_flushed;         /* 1 if some bitmap is not flushed */
-  my_bool flush_all_requested;         /**< If _ma_bitmap_flush_all waiting */
+  uint flush_all_requested;            /**< If _ma_bitmap_flush_all waiting */
   uint non_flushable;                  /**< 0 if bitmap and log are in sync */
   PAGECACHE_FILE file;		       /* datafile where bitmap is stored */