Merge bk-internal:/home/bk/mysql-maria

into  mysql.com:/data0/mysqldev/my/build-200802121236-5.1.23a-maria-alpha/mysql-5.1-release

KNOWN_BUGS.txt

 This file should contain all know fatal bugs in the Maria storage
 engine for the last source or binary release.  Minor bugs, extensions
 and feature request and bugs found since this release can be find in the
-MySQL bugs databases at: http://bugs.mysql.com/
+MySQL bugs databases at: http://bugs.mysql.com/ (category "Maria
+storage engine").
 
 There shouldn't normally be any bugs that affects normal operations in
 any Maria release.  Still, there are always exceptions and edge cases
@@ -37,9 +38,9 @@ Known bugs that we are working on and will be fixed shortly
 - LOAD INDEX commands are for the moment ignored for Maria tables
   (The code needs to be rewritten to do all reads through page cache to
   avoid half-block reads)
-- Freeing maria tmp table after fetching rows using prepared statements causes
-  a crash.
-
+- Some concurrency bugs in Maria's page cache which sometimes show up
+  under load http://bugs.mysql.com/bug.php?id=34161 and
+  http://bugs.mysql.com/bug.php?id=34634 .
 
 Known bugs that are planned to be fixed before Beta
 ===================================================

include/my_pthread.h

@@ -674,6 +674,9 @@ extern int pthread_dummy(int);
 #endif
 #endif
 
+#define MY_PTHREAD_LOCK_READ 0
+#define MY_PTHREAD_LOCK_WRITE 1
+
 struct st_my_thread_var
 {
   int thr_errno;
@@ -688,6 +691,7 @@ struct st_my_thread_var
   my_bool init;
   struct st_my_thread_var *next,**prev;
   void *opt_info;
+  uint  lock_type; /* used by conditional release the queue */
 #ifndef DBUG_OFF
   void *dbug;
   char name[THREAD_NAME_SIZE+1];

include/wqueue.h

@@ -20,6 +20,7 @@ void wqueue_add_and_wait(WQUEUE *wqueue,
                          struct st_my_thread_var *thread,
                          pthread_mutex_t *lock);
 void wqueue_release_queue(WQUEUE *wqueue);
+void wqueue_release_one_locktype_from_queue(WQUEUE *wqueue);
 
 #endif
 

mysys/wqueue.c

@@ -136,6 +136,62 @@ void wqueue_release_queue(WQUEUE *wqueue)
 }
 
 
+/**
+  @brief Removes all threads waiting for read or first one waiting for write.
+
+  @param wqueue          pointer to the queue structure
+  @apram thread          pointer to the thread to be added to the queue
+*/
+
+void wqueue_release_one_locktype_from_queue(WQUEUE *wqueue)
+{
+  struct st_my_thread_var *last= wqueue->last_thread;
+  struct st_my_thread_var *next= last->next;
+  struct st_my_thread_var **prev= &last->next;
+  struct st_my_thread_var *thread;
+  struct st_my_thread_var *new_last= NULL;
+  uint first_type= next->lock_type;
+  if (first_type == MY_PTHREAD_LOCK_WRITE)
+  {
+    /* release first waiting for write lock */
+    thread= next;
+    pthread_cond_signal(&thread->suspend);
+    if (thread == last)
+      wqueue->last_thread= NULL;
+    *prev= thread->next;
+    thread->next= NULL;
+    return;
+  }
+  do
+  {
+    thread= next;
+    next= thread->next;
+    if (thread->lock_type == MY_PTHREAD_LOCK_WRITE)
+    {
+      /* skip waiting for write lock */
+      *prev= thread;
+      prev= &thread->next;
+      new_last= NULL;
+    }
+    else
+    {
+      /* release waiting for read lock */
+      pthread_cond_signal(&thread->suspend);
+      new_last= thread->next;
+      thread->next= NULL;
+    }
+  } while (thread != last);
+  if (new_last)
+  {
+    /* last was deleted */
+    if (new_last == last)
+      wqueue->last_thread= NULL; /* empty list */
+    else
+      wqueue->last_thread= new_last;
+  }
+}
+
+
 /*
   Add thread and wait
 

sql/sql_repl.cc

@@ -1522,7 +1522,7 @@ bool show_binlogs(THD* thd)
   if (!mysql_bin_log.is_open())
   {
     my_message(ER_NO_BINARY_LOGGING, ER(ER_NO_BINARY_LOGGING), MYF(0));
-    return 1;
+    DBUG_RETURN(TRUE);
   }
 
   field_list.push_back(new Item_empty_string("Log_name", 255));

storage/maria/CMakeLists.txt

@@ -62,6 +62,10 @@ TARGET_LINK_LIBRARIES(maria_read_log maria myisam mysys dbug strings zlib wsock3
 ADD_EXECUTABLE(maria_pack maria_pack.c)
 TARGET_LINK_LIBRARIES(maria_pack maria myisam mysys dbug strings zlib wsock32)
 
+ADD_EXECUTABLE(maria_dump_log ma_loghandler.c)
+TARGET_LINK_LIBRARIES(maria_dump_log maria myisam mysys dbug strings zlib wsock32)
+SET_TARGET_PROPERTIES(maria_dump_log PROPERTIES COMPILE_FLAGS "-DMARIA_DUMP_LOG")
+
 ADD_EXECUTABLE(ma_test1 ma_test1.c)
 TARGET_LINK_LIBRARIES(ma_test1 maria myisam mysys dbug strings zlib wsock32)
 

storage/maria/ma_blockrec.c

@@ -6343,12 +6343,13 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
         length= data_size;
         if (i == 0 && sub_ranges == 0)
         {
-          /* Last page may be only partly filled. */
+          /*
+            Last page may be only partly filled. We zero the rest, like
+            write_full_pages() does.
+          */
           length-= empty_space;
-#ifdef IDENTICAL_PAGES_AFTER_RECOVERY
           bzero(buff + share->block_size - PAGE_SUFFIX_SIZE - empty_space,
                 empty_space);
-#endif
         }
         memcpy(buff+ PAGE_TYPE_OFFSET + 1, data, length);
         data+= length;

storage/maria/ma_check.c

@@ -2983,11 +2983,9 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
     if (zero_lsn)
       bzero(buff, LSN_SIZE);
     length= _ma_get_page_used(share, buff);
-    /* Skip mailformed blocks */
-    DBUG_ASSERT(length + share->keypage_header <= block_size);
-    if (length + share->keypage_header < block_size)
-      bzero(buff + share->keypage_header + length, block_size - length -
-            share->keypage_header);
+    DBUG_ASSERT(length <= block_size);
+    if (length < block_size)
+      bzero(buff + length, block_size - length);
     pagecache_unlock_by_link(share->pagecache, page_link.link,
                              PAGECACHE_LOCK_WRITE_UNLOCK,
                              PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
@@ -3001,7 +2999,7 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
 
 
 /**
-   @brief Fill empty space in index file with zeroes
+   @brief Fill empty space in data file with zeroes
 
    @todo
    Zerofill all pages marked in bitmap as empty and change them to

storage/maria/ma_loghandler.c

@@ -7475,7 +7475,10 @@ int translog_assign_id_to_share(MARIA_HA *tbl_info, TRN *trn)
                                                   1].length),
                                        sizeof(log_array)/sizeof(log_array[0]),
                                        log_array, log_data, NULL)))
+    {
+      pthread_mutex_unlock(&share->intern_lock);
       return 1;
+    }
   }
   pthread_mutex_unlock(&share->intern_lock);
   return 0;
@@ -8381,7 +8384,7 @@ int main(int argc, char **argv)
        opt_offset+= TRANSLOG_PAGE_SIZE, opt_pages--)
   {
     if (my_pread(handler, buffer, TRANSLOG_PAGE_SIZE, opt_offset,
-                 MYF(MY_FNABP | MY_WME)))
+                 MYF(MY_NABP)))
     {
       if (my_errno == HA_ERR_FILE_TOO_SHORT)
         goto end;

storage/maria/ma_pagecache.h

@@ -300,7 +300,7 @@ extern my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
 extern int reset_pagecache_counters(const char *name, PAGECACHE *pagecache);
 extern uchar *pagecache_block_link_to_buffer(PAGECACHE_BLOCK_LINK *block);
 
-extern uint pagacache_pagelevel(PAGECACHE_BLOCK_LINK *block);
+extern uint pagecache_pagelevel(PAGECACHE_BLOCK_LINK *block);
 extern void pagecache_add_level_by_link(PAGECACHE_BLOCK_LINK *block,
 					uint level);
 

storage/maria/maria_chk.c

@@ -375,12 +375,10 @@ static struct my_option my_long_options[] =
   { "zerofill", 'z',
     "Fill empty space in data and index files with zeroes",
     0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
-#ifdef IDENTICAL_PAGES_AFTER_RECOVERY
   { "zerofill-keep-lsn", OPT_ZEROFILL_KEEP_LSN,
     "Like --zerofill but does not zero out LSN of data/index pages;"
     " used only for testing and debugging",
     0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
-#endif
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 
@@ -511,13 +509,9 @@ static void usage(void)
 		      (It may be VERY slow to do a sort the first time!).\n\
   -b,  --block-search=#\n\
                       Find a record, a block at given offset belongs to.\n\
-  -z,  --zerofill     Fill empty space in data and index files with zeroes"
-#ifdef IDENTICAL_PAGES_AFTER_RECOVERY
-"\n\
+  -z,  --zerofill     Fill empty space in data and index files with zeroes\n\
   --zerofill-keep-lsn Like --zerofill but does not zero out LSN of\n\
-                      data/index pages; used only for testing and debugging"
-#endif
-       ".");
+                      data/index pages.");
 
   print_defaults("my", load_default_groups);
   my_print_variables(my_long_options);
@@ -790,14 +784,12 @@ get_one_option(int optid,
     else
       check_param.testflag|= T_ZEROFILL;
     break;
-#ifdef IDENTICAL_PAGES_AFTER_RECOVERY
   case OPT_ZEROFILL_KEEP_LSN:
     if (argument == disabled_my_option)
       check_param.testflag&= ~(T_ZEROFILL_KEEP_LSN | T_ZEROFILL);
     else
       check_param.testflag|= (T_ZEROFILL_KEEP_LSN | T_ZEROFILL);
     break;
-#endif
   case 'H':
     my_print_help(my_long_options);
     exit(0);
@@ -1123,13 +1115,23 @@ static int maria_chk(HA_CHECK *param, char *filename)
     if (!error)
     {
       /*
-        Tell the server's Recovery to ignore old REDOs on this table; we don't
+        Unless this was only --zerofill-keep-lsn, old REDOs are not
+        applicable, tell the server's Recovery to ignore them; we don't
         know what the log's end LSN is now, so we just let the server know
         that it will have to find and store it.
         This is the only case where create_rename_lsn can be a horizon and not
         a LSN.
+        If this was only --zerofill-keep-lsn, the table can be used in
+        Recovery and especially in this scenario: do a dirty-copy-based backup
+        (snapshot-like), --zerofill-keep-lsn on the copies to achieve better
+        compression, compress the copies with an external tool, and after a
+        restore, Recovery still works (because pages and state still have
+        their correct LSNs).
       */
-      if (share->base.born_transactional)
+      if (share->base.born_transactional &&
+          ((param->testflag & (T_REP_ANY | T_SORT_RECORDS | T_SORT_INDEX |
+                               T_ZEROFILL | T_ZEROFILL_KEEP_LSN)) !=
+           (T_ZEROFILL | T_ZEROFILL_KEEP_LSN)))
         share->state.create_rename_lsn= share->state.is_of_horizon=
           share->state.skip_redo_lsn= LSN_REPAIRED_BY_MARIA_CHK;
     }

storage/maria/unittest/Makefile.am

@@ -39,6 +39,7 @@ noinst_PROGRAMS =	ma_control_file-t trnman-t lockman2-t \
 			ma_pagecache_consist_64kRD-t \
 			ma_pagecache_consist_1kWR-t \
 			ma_pagecache_consist_64kWR-t \
+                        ma_pagecache_rwconsist_1k-t \
 			ma_test_loghandler-t \
                         ma_test_loghandler_multigroup-t \
 			ma_test_loghandler_multithread-t \
@@ -96,6 +97,9 @@ ma_pagecache_consist_1kWR_t_CPPFLAGS = 	$(ma_pagecache_common_cppflags) -DTEST_P
 ma_pagecache_consist_64kWR_t_SOURCES =	$(ma_pagecache_consist_src)
 ma_pagecache_consist_64kWR_t_CPPFLAGS =	$(ma_pagecache_common_cppflags) -DTEST_PAGE_SIZE=65536 -DTEST_WRITERS
 
+ma_pagecache_rwconsist_1k_t_SOURCES =	ma_pagecache_rwconsist.c
+ma_pagecache_rwconsist_1k_t_CPPFLAGS = -DTEST_PAGE_SIZE=1024
+
 # the generic lock manager may not be used in the end and lockman1-t crashes,
 # so we don't build lockman-t and lockman1-t 
 CLEANFILES =		maria_log_control page_cache_test_file_1 \

storage/maria/unittest/ma_control_file-t.c

@@ -104,7 +104,7 @@ int (*default_error_handler_hook)(uint my_err, const char *str,
 
 /* like ma_control_file_create_or_open(), but without error messages */
 
-static CONTROL_FILE_ERROR local_ma_control_file_create_or_open()
+static CONTROL_FILE_ERROR local_ma_control_file_create_or_open(void)
 {
   CONTROL_FILE_ERROR error;
   error_handler_hook= my_ignore_message;
@@ -205,7 +205,7 @@ static int close_file()
 
 static int create_or_open_file()
 {
-  RET_ERR_UNLESS(local_ma_control_file_create_or_open(TRUE) == CONTROL_FILE_OK);
+  RET_ERR_UNLESS(local_ma_control_file_create_or_open() == CONTROL_FILE_OK);
   /* Check that the module reports expected information */
   RET_ERR_UNLESS(verify_module_values_match_expected() == 0);
   return 0;
@@ -362,7 +362,7 @@ static int test_bad_magic_string()
                            MYF(MY_FNABP |  MY_WME)) == 0);
 
   /* Check that control file module sees the problem */
-  RET_ERR_UNLESS(local_ma_control_file_create_or_open(TRUE) ==
+  RET_ERR_UNLESS(local_ma_control_file_create_or_open() ==
              CONTROL_FILE_BAD_MAGIC_STRING);
   /* Restore magic string */
   RET_ERR_UNLESS(my_pwrite(fd, buffer, 4, 0, MYF(MY_FNABP |  MY_WME)) == 0);
@@ -388,7 +388,7 @@ static int test_bad_checksum()
   buffer[0]+= 3; /* mangle checksum */
   RET_ERR_UNLESS(my_pwrite(fd, buffer, 1, 30, MYF(MY_FNABP |  MY_WME)) == 0);
   /* Check that control file module sees the problem */
-  RET_ERR_UNLESS(local_ma_control_file_create_or_open(TRUE) ==
+  RET_ERR_UNLESS(local_ma_control_file_create_or_open() ==
                  CONTROL_FILE_BAD_CHECKSUM);
   /* Restore checksum */
   buffer[0]-= 3;
@@ -403,7 +403,7 @@ static int test_bad_blocksize()
 {
   maria_block_size<<= 1;
   /* Check that control file module sees the problem */
-  RET_ERR_UNLESS(local_ma_control_file_create_or_open(TRUE) ==
+  RET_ERR_UNLESS(local_ma_control_file_create_or_open() ==
                  CONTROL_FILE_WRONG_BLOCKSIZE);
   /* Restore blocksize */
   maria_block_size>>= 1;
@@ -478,7 +478,7 @@ static int test_bad_hchecksum()
   buffer[0]+= 3; /* mangle checksum */
   RET_ERR_UNLESS(my_pwrite(fd, buffer, 1, 26, MYF(MY_FNABP |  MY_WME)) == 0);
   /* Check that control file module sees the problem */
-  RET_ERR_UNLESS(local_ma_control_file_create_or_open(TRUE) ==
+  RET_ERR_UNLESS(local_ma_control_file_create_or_open() ==
                  CONTROL_FILE_BAD_HEAD_CHECKSUM);
   /* Restore checksum */
   buffer[0]-= 3;
@@ -502,14 +502,14 @@ static int test_bad_size()
                           MYF(MY_WME))) >= 0);
   RET_ERR_UNLESS(my_write(fd, buffer, 10, MYF(MY_FNABP |  MY_WME)) == 0);
   /* Check that control file module sees the problem */
-  RET_ERR_UNLESS(local_ma_control_file_create_or_open(TRUE) ==
+  RET_ERR_UNLESS(local_ma_control_file_create_or_open() ==
                  CONTROL_FILE_TOO_SMALL);
   for (i= 0; i < 8; i++)
   {
     RET_ERR_UNLESS(my_write(fd, buffer, 66, MYF(MY_FNABP |  MY_WME)) == 0);
   }
   /* Check that control file module sees the problem */
-  RET_ERR_UNLESS(local_ma_control_file_create_or_open(TRUE) ==
+  RET_ERR_UNLESS(local_ma_control_file_create_or_open() ==
                  CONTROL_FILE_TOO_BIG);
   RET_ERR_UNLESS(my_close(fd, MYF(MY_WME)) == 0);
 

storage/maria/unittest/ma_pagecache_consist.c

@@ -273,46 +273,49 @@ void writer(int num)
 static void *test_thread_reader(void *arg)
 {
   int param=*((int*) arg);
-  DBUG_ENTER("test_reader");
-
   my_thread_init();
-
-  DBUG_PRINT("enter", ("param: %d", param));
-
-  reader(param);
-
-  DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
-  pthread_mutex_lock(&LOCK_thread_count);
-  ok(1, "reader%d: done", param);
-  thread_count--;
-  VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
-  pthread_mutex_unlock(&LOCK_thread_count);
-  free((uchar*) arg);
-  my_thread_end();
-  DBUG_RETURN(0);
+  {
+    DBUG_ENTER("test_reader");
+    DBUG_PRINT("enter", ("param: %d", param));
+
+    reader(param);
+
+    DBUG_PRINT("info", ("Thread %s ended", my_thread_name()));
+    pthread_mutex_lock(&LOCK_thread_count);
+    ok(1, "reader%d: done", param);
+    thread_count--;
+    VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
+    pthread_mutex_unlock(&LOCK_thread_count);
+    free((uchar*) arg);
+    my_thread_end();
+  }
+  return 0;
 }
 
+
 static void *test_thread_writer(void *arg)
 {
   int param=*((int*) arg);
-  DBUG_ENTER("test_writer");
-
   my_thread_init();
-  DBUG_PRINT("enter", ("param: %d", param));
-
-  writer(param);
-
-  DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
-  pthread_mutex_lock(&LOCK_thread_count);
-  ok(1, "writer%d: done", param);
-  thread_count--;
-  VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
-  pthread_mutex_unlock(&LOCK_thread_count);
-  free((uchar*) arg);
-  my_thread_end();
-  DBUG_RETURN(0);
+  {
+    DBUG_ENTER("test_writer");
+    DBUG_PRINT("enter", ("param: %d", param));
+
+    writer(param);
+
+    DBUG_PRINT("info", ("Thread %s ended", my_thread_name()));
+    pthread_mutex_lock(&LOCK_thread_count);
+    ok(1, "writer%d: done", param);
+    thread_count--;
+    VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
+    pthread_mutex_unlock(&LOCK_thread_count);
+    free((uchar*) arg);
+    my_thread_end();
+  }
+  return 0;
 }
 
+
 int main(int argc __attribute__((unused)),
          char **argv __attribute__((unused)))
 {

storage/maria/unittest/ma_pagecache_rwconsist.c

+/*
+  TODO: use pthread_join instead of wait_for_thread_count_to_be_zero, like in
+  my_atomic-t.c (see BUG#22320).
+*/
+
+#include <tap.h>
+#include <my_sys.h>
+#include <m_string.h>
+#include "test_file.h"
+#include <tap.h>
+
+#define PCACHE_SIZE (TEST_PAGE_SIZE*1024*8)
+
+#ifndef DBUG_OFF
+static const char* default_dbug_option;
+#endif
+
+
+#define SLEEP usleep(5)
+
+static char *file1_name= (char*)"page_cache_test_file_1";
+static PAGECACHE_FILE file1;
+static pthread_cond_t COND_thread_count;
+static pthread_mutex_t LOCK_thread_count;
+static uint thread_count;
+static PAGECACHE pagecache;
+
+static uint number_of_readers= 5;
+static uint number_of_writers= 5;
+static uint number_of_read_tests= 2000;
+static uint number_of_write_tests= 1000;
+static uint read_sleep_limit= 3;
+static uint report_divisor= 50;
+
+/**
+  @brief Dummy pagecache callback.
+*/
+
+static my_bool
+dummy_callback(uchar *page __attribute__((unused)),
+               pgcache_page_no_t page_no __attribute__((unused)),
+               uchar* data_ptr __attribute__((unused)))
+{
+  return 0;
+}
+
+
+/**
+  @brief Dummy pagecache callback.
+*/
+
+static void
+dummy_fail_callback(uchar* data_ptr __attribute__((unused)))
+{
+  return;
+}
+
+
+/**
+  @brief Checks page consistency
+
+  @param buff            pointer to the page content
+  @param task            task ID
+*/
+void check_page(uchar *buff, int task)
+{
+  uint i;
+  DBUG_ENTER("check_page");
+
+  for (i= 1; i < TEST_PAGE_SIZE; i++)
+  {
+    if (buff[0] != buff[i])
+      goto err;
+  }
+  DBUG_VOID_RETURN;
+err:
+  diag("Task %d char #%u '%u' != '%u'", task, i, (uint) buff[0],
+       (uint) buff[i]);
+  DBUG_PRINT("err", ("try to flush"));
+  exit(1);
+}
+
+
+
+void reader(int num)
+{
+  unsigned char *buff;
+  uint i;
+  PAGECACHE_BLOCK_LINK *link;
+
+  for (i= 0; i < number_of_read_tests; i++)
+  {
+    if (i % report_divisor == 0)
+      diag("Reader %d - %u", num, i);
+    buff= pagecache_read(&pagecache, &file1, 0, 3, NULL,
+                         PAGECACHE_PLAIN_PAGE,
+                         PAGECACHE_LOCK_READ,
+                         &link);
+    check_page(buff, num);
+    pagecache_unlock_by_link(&pagecache, link,
+                             PAGECACHE_LOCK_READ_UNLOCK,
+                             PAGECACHE_UNPIN, 0, 0, 0);
+    {
+      int lim= rand() % read_sleep_limit;
+      int j;
+      for (j= 0; j < lim; j++)
+        SLEEP;
+    }
+  }
+}
+
+
+void writer(int num)
+{
+  uint i;
+  uchar *buff;
+  PAGECACHE_BLOCK_LINK *link;
+
+  for (i= 0; i < number_of_write_tests; i++)
+  {
+    uchar c= (uchar) rand() % 256;
+
+    if (i % report_divisor == 0)
+      diag("Writer %d - %u", num, i);
+    buff= pagecache_read(&pagecache, &file1, 0, 3, NULL,
+                         PAGECACHE_PLAIN_PAGE,
+                         PAGECACHE_LOCK_WRITE,
+                         &link);
+
+    check_page(buff, num);
+    bfill(buff, TEST_PAGE_SIZE / 2, c);
+    SLEEP;
+    bfill(buff + TEST_PAGE_SIZE/2, TEST_PAGE_SIZE / 2, c);
+    check_page(buff, num);
+    pagecache_unlock_by_link(&pagecache, link,
+                             PAGECACHE_LOCK_WRITE_UNLOCK,
+                             PAGECACHE_UNPIN, 0, 0, 1);
+    SLEEP;
+  }
+}
+
+
+static void *test_thread_reader(void *arg)
+{
+  int param=*((int*) arg);
+  my_thread_init();
+  {
+    DBUG_ENTER("test_reader");
+
+    DBUG_PRINT("enter", ("param: %d", param));
+
+    reader(param);
+
+    DBUG_PRINT("info", ("Thread %s ended", my_thread_name()));
+    pthread_mutex_lock(&LOCK_thread_count);
+    ok(1, "reader%d: done", param);
+    thread_count--;
+    VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
+    pthread_mutex_unlock(&LOCK_thread_count);
+    free((uchar*) arg);
+    my_thread_end();
+  }
+  return 0;
+}
+
+
+static void *test_thread_writer(void *arg)
+{
+  int param=*((int*) arg);
+  my_thread_init();
+  {
+    DBUG_ENTER("test_writer");
+
+    writer(param);
+
+    DBUG_PRINT("info", ("Thread %s ended", my_thread_name()));
+    pthread_mutex_lock(&LOCK_thread_count);
+    ok(1, "writer%d: done", param);
+    thread_count--;
+    VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
+    pthread_mutex_unlock(&LOCK_thread_count);
+    free((uchar*) arg);
+    my_thread_end();
+  }
+  return 0;
+}
+
+
+int main(int argc __attribute__((unused)),
+         char **argv __attribute__((unused)))
+{
+  pthread_t tid;
+  pthread_attr_t thr_attr;
+  int *param, error, pagen;
+
+  MY_INIT(argv[0]);
+
+#ifndef DBUG_OFF
+#if defined(__WIN__)
+  default_dbug_option= "d:t:i:O,\\test_pagecache_consist.trace";
+#else
+  default_dbug_option= "d:t:i:o,/tmp/test_pagecache_consist.trace";
+#endif
+  if (argc > 1)
+  {
+    DBUG_SET(default_dbug_option);
+    DBUG_SET_INITIAL(default_dbug_option);
+  }
+#endif
+
+  {
+  DBUG_ENTER("main");
+  DBUG_PRINT("info", ("Main thread: %s\n", my_thread_name()));
+  plan(number_of_writers + number_of_readers);
+  SKIP_BIG_TESTS(number_of_writers + number_of_readers)
+  {
+
+  if ((file1.file= my_open(file1_name,
+                           O_CREAT | O_TRUNC | O_RDWR, MYF(0))) == -1)
+  {
+    diag( "Got error during file1 creation from open() (errno: %d)\n",
+	    errno);
+    exit(1);
+  }
+  pagecache_file_init(file1, &dummy_callback, &dummy_callback,
+                      &dummy_fail_callback, &dummy_callback, NULL);
+  DBUG_PRINT("info", ("file1: %d", file1.file));
+  if (my_chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO, MYF(MY_WME)))
+    exit(1);
+  my_pwrite(file1.file, (const uchar*) "test file", 9, 0, MYF(0));
+
+  if ((error= pthread_cond_init(&COND_thread_count, NULL)))
+  {
+    diag( "COND_thread_count: %d from pthread_cond_init (errno: %d)\n",
+	    error, errno);
+    exit(1);
+  }
+  if ((error= pthread_mutex_init(&LOCK_thread_count, MY_MUTEX_INIT_FAST)))
+  {
+    diag( "LOCK_thread_count: %d from pthread_cond_init (errno: %d)\n",
+	    error, errno);
+    exit(1);
+  }
+
+  if ((error= pthread_attr_init(&thr_attr)))
+  {
+    diag("Got error: %d from pthread_attr_init (errno: %d)\n",
+	    error,errno);
+    exit(1);
+  }
+  if ((error= pthread_attr_setdetachstate(&thr_attr, PTHREAD_CREATE_DETACHED)))
+  {
+    diag(
+	    "Got error: %d from pthread_attr_setdetachstate (errno: %d)\n",
+	    error,errno);
+    exit(1);
+  }
+
+#ifdef HAVE_THR_SETCONCURRENCY
+  VOID(thr_setconcurrency(2));
+#endif
+
+  if ((pagen= init_pagecache(&pagecache, PCACHE_SIZE, 0, 0,
+                             TEST_PAGE_SIZE, 0)) == 0)
+  {
+    diag("Got error: init_pagecache() (errno: %d)\n",
+            errno);
+    exit(1);
+  }
+  DBUG_PRINT("info", ("Page cache %d pages", pagen));
+  {
+    unsigned char *buffr= malloc(TEST_PAGE_SIZE);
+    memset(buffr, '\0', TEST_PAGE_SIZE);
+    pagecache_write(&pagecache, &file1, 0, 3, buffr,
+                    PAGECACHE_PLAIN_PAGE,
+                    PAGECACHE_LOCK_LEFT_UNLOCKED,
+                    PAGECACHE_PIN_LEFT_UNPINNED,
+                    PAGECACHE_WRITE_DELAY,
+                    0, LSN_IMPOSSIBLE);
+  }
+  pthread_mutex_lock(&LOCK_thread_count);
+
+  while (number_of_readers != 0 || number_of_writers != 0)
+  {
+    if (number_of_readers != 0)
+    {
+      param=(int*) malloc(sizeof(int));
+      *param= number_of_readers + number_of_writers;
+      if ((error= pthread_create(&tid, &thr_attr, test_thread_reader,
+                                 (void*) param)))
+      {
+        diag("Got error: %d from pthread_create (errno: %d)\n",
+                error,errno);
+        exit(1);
+      }
+      thread_count++;
+      number_of_readers--;
+    }
+    if (number_of_writers != 0)
+    {
+      param=(int*) malloc(sizeof(int));
+      *param= number_of_writers + number_of_readers;
+      if ((error= pthread_create(&tid, &thr_attr, test_thread_writer,
+                                 (void*) param)))
+      {
+        diag("Got error: %d from pthread_create (errno: %d)\n",
+                error,errno);
+        exit(1);
+      }
+      thread_count++;
+      number_of_writers--;
+    }
+  }
+  DBUG_PRINT("info", ("Thread started"));
+  pthread_mutex_unlock(&LOCK_thread_count);
+
+  pthread_attr_destroy(&thr_attr);
+
+  /* wait finishing */
+  pthread_mutex_lock(&LOCK_thread_count);
+  while (thread_count)
+  {
+    if ((error= pthread_cond_wait(&COND_thread_count,&LOCK_thread_count)))
+      diag("COND_thread_count: %d from pthread_cond_wait\n",error);
+  }
+  pthread_mutex_unlock(&LOCK_thread_count);
+  DBUG_PRINT("info", ("thread ended"));
+
+  end_pagecache(&pagecache, 1);
+  DBUG_PRINT("info", ("Page cache ended"));
+
+  if (my_close(file1.file, MYF(0)) != 0)
+  {
+    diag( "Got error during file1 closing from close() (errno: %d)\n",
+	    errno);
+    exit(1);
+  }
+  my_delete(file1_name, MYF(0));
+
+  DBUG_PRINT("info", ("file1 (%d) closed", file1.file));
+  DBUG_PRINT("info", ("Program end"));
+
+  } /* SKIP_BIG_TESTS */
+  my_end(0);
+
+  return exit_status();
+  }
+}

storage/maria/unittest/ma_test_recovery.pl

@@ -2,6 +2,7 @@
 
 use Getopt::Long;
 use File::Copy;
+use File::Compare;
 use File::Basename;
 
 $|= 1;
@@ -17,6 +18,7 @@ my $tmp= "./tmp";
 my $my_progname= $0;
 my $suffix;
 my $md5sum;
+my $zerofilled_tables= 0;
 
 $my_progname=~ s/.*[\/]//;
 $maria_path= dirname($0) . "/..";
@@ -89,18 +91,6 @@ sub main
     mkdir $tmp;
   }
   print "MARIA RECOVERY TESTS\n";
-  $res= `$maria_exe_path/maria_read_log$suffix --help | grep IDENTICAL_PAGES_AFTER_RECOVERY`;
-
-  if (length($res))
-  {
-    print "Recovery tests require compilation with DBUG\n";
-    print "Aborting test\n";
-    # In the future, we will not abort but use maria_chk --zerofill-keep-lsn
-    # for comparisons in non-debug builds.
-    # For now we just skip the test, pretending it passed (nothing is
-    # alarming).
-    exit(0);
-  }
 
   # To not flood the screen, we redirect all the commands below to a text file
   # and just give a final error if their output is not as expected
@@ -140,18 +130,14 @@ sub main
     move("$table.MAI", "$tmp/$table-good.MAI") ||
       die "Can't move $table.MAI to $tmp/$table-good.MAI\n";
     apply_log($table, "shouldnotchangelog");
-    $res= `cmp $table.MAD $tmp/$table-good.MAD`;
-    print MY_LOG $res;
-    $res= `cmp $table.MAI $tmp/$table-good.MAI`;
-    print MY_LOG $res;
     check_table_is_same($table, $checksum);
+    $res= physical_cmp($table, "$tmp/$table-good");
+    print MY_LOG $res;
     print MY_LOG "testing idempotency\n";
     apply_log($table, "shouldnotchangelog");
-    $res= `cmp $table.MAD $tmp/$table-good.MAD`;
-    print MY_LOG $res;
-    $res= `cmp $table.MAI $tmp/$table-good.MAI`;
-    print MY_LOG $res;
     check_table_is_same($table, $checksum);
+    $res= physical_cmp($table, "$tmp/$table-good");
+    print MY_LOG $res;
   }
 
   print MY_LOG "Testing the REDO AND UNDO PHASE\n";
@@ -255,21 +241,16 @@ sub main
           check_table_is_same($table, $checksum);
           print MY_LOG "testing idempotency\n";
           apply_log($table, "shouldnotchangelog");
-          # We can't do a binary compary as there may have been different number
-          # of calls to compact_page. We can enable this if we first call
-          # maria-check to generate identically compacted pages.
-          #    cmp $table.MAD $tmp/$table-after_undo.MAD
-          $res= `cmp $table.MAI $tmp/$table-after_undo.MAI`;
-          print MY_LOG $res;
           check_table_is_same($table, $checksum);
+          $res= physical_cmp($table, "$tmp/$table-after_undo");
+          print MY_LOG $res;
           print MY_LOG "testing applying of CLRs to recreate table\n";
           unlink <$table.MA?>;
           #    cp $tmp/maria_log* $maria_path  #unneeded
           apply_log($table, "shouldnotchangelog");
-          #    cmp $table.MAD $tmp/$table-after_undo.MAD
-          $res= `cmp $table.MAI $tmp/$table-after_undo.MAI`;
-          print MY_LOG $res;
           check_table_is_same($table, $checksum);
+          $res= physical_cmp($table, "$tmp/$table-after_undo");
+          print MY_LOG $res;
         }
         unlink <$table.* $tmp/$table* $tmp/maria_chk_*.txt $tmp/maria_read_log_$table.txt>;
       }
@@ -286,12 +267,13 @@ sub main
   # does not put back the "analyzed,optimized keys"(etc) index state.
   `diff -b $maria_path/unittest/ma_test_recovery.expected $tmp/ma_test_recovery.output`;
   if ($? >> 8) {
-    print "UNEXPECTED OUTPUT OF TESTS, FAILED\n";
+    print "UNEXPECTED OUTPUT OF TESTS, FAILED";
+    print " (zerofilled $zerofilled_tables tables)\n";
     print "For more info, do diff -b $maria_path/unittest/ma_test_recovery.expected ";
     print "$tmp/ma_test_recovery.output\n";
     exit(1);
   }
-  print "ALL RECOVERY TESTS OK\n";
+  print "ALL RECOVERY TESTS OK (zerofilled $zerofilled_tables tables)\n";
 }
 
 ####
@@ -357,7 +339,7 @@ sub apply_log
   {
     print MY_LOG "bad argument '$shouldchangelog'\n";
     return 1;
-  }
+  } 
   $log_md5= `$md5sum maria_log.*`;
 
   print MY_LOG "applying log\n";
@@ -395,6 +377,58 @@ sub my_which
 }
 
 
+####
+#### physical_cmp: compares two tables (MAI and MAD) physically;
+#### uses zerofill-keep-lsn to reduce irrelevant differences.
+####
+
+sub physical_cmp
+{
+  my ($table1, $table2)= @_;
+  my ($zerofilled, $ret_text)= (0, "");
+  #return `cmp $table1.MAD $table2.MAD`.`cmp $table1.MAI $table2.MAI`;
+  foreach my $file_suffix ("MAD", "MAI")
+  {
+    my $file1= "$table1.$file_suffix";
+    my $file2= "$table2.$file_suffix";
+    my $res= File::Compare::compare($file1, $file2);
+    die() if ($res == -1);
+    if ($res == 1 # they differ
+        and !$zerofilled)
+    {
+      # let's try with --zerofill-keep-lsn
+      $zerofilled= 1; # but no need to do it twice
+      $zerofilled_tables= $zerofilled_tables + 1;
+      my $table_no= 1;
+      foreach my $table ($table1, $table2)
+      {
+        # save original tables to restore them later
+        copy("$table.MAD", "$tmp/before_zerofill$table_no.MAD") || die();
+        copy("$table.MAI", "$tmp/before_zerofill$table_no.MAI") || die();
+        $com= "$maria_exe_path/maria_chk$suffix -s --zerofill-keep-lsn $table";
+        $res= `$com`;
+        print MY_LOG $res;
+        $table_no= $table_no + 1;
+      }
+      $res= File::Compare::compare($file1, $file2);
+      die() if ($res == -1);
+    }
+    $ret_text.= "$file1 and $file2 differ\n" if ($res != 0);
+  }
+  if ($zerofilled)
+  {
+    my $table_no= 1;
+    foreach my $table ($table1, $table2)
+    {
+      move("$tmp/before_zerofill$table_no.MAD", "$table.MAD") || die();
+      move("$tmp/before_zerofill$table_no.MAI", "$table.MAI") || die();
+      $table_no= $table_no + 1;
+    }
+  }
+  return $ret_text;
+}
+
+
 ####
 #### usage
 ####