Write failure callback added.

storage/maria/ma_key_recover.c:
  Initialization fixed (by Monty)

storage/maria/ma_bitmap.c

@@ -222,6 +222,7 @@ my_bool _ma_bitmap_init(MARIA_SHARE *share, File file)
   bitmap->block_size= share->block_size;
   bitmap->file.file= file;
   bitmap->file.callback_data= (uchar*) share;
+  bitmap->file.write_fail= &maria_page_write_failure;
   if (share->temporary)
   {
     bitmap->file.read_callback=  &maria_page_crc_check_none;

storage/maria/ma_check.c

@@ -5604,11 +5604,13 @@ my_bool create_new_data_handle(MARIA_SORT_PARAM *param, File new_file)
   pagecache_file_init(new_info->s->bitmap.file, &maria_page_crc_check_bitmap,
                       (new_info->s->options & HA_OPTION_PAGE_CHECKSUM ?
                        &maria_page_crc_set_normal :
-                       &maria_page_filler_set_bitmap), new_info->s);
+                       &maria_page_filler_set_bitmap),
+                      &maria_page_write_failure, new_info->s);
   pagecache_file_init(new_info->dfile, &maria_page_crc_check_data,
                       (new_info->s->options & HA_OPTION_PAGE_CHECKSUM ?
                        &maria_page_crc_set_normal :
-                       &maria_page_filler_set_normal), new_info->s);
+                       &maria_page_filler_set_normal),
+                      &maria_page_write_failure, new_info->s);
   change_data_file_descriptor(new_info, new_file);
   maria_lock_database(new_info, F_EXTRA_LCK);
   if ((sort_info->param->testflag & T_UNPACK) &&

storage/maria/ma_key_recover.c

@@ -603,6 +603,7 @@ uint _ma_apply_redo_index_new_page(MARIA_HA *info, LSN lsn,
   }
 
   /* Write modified page */
+  bzero(buff, LSN_STORE_SIZE);
   memcpy(buff + LSN_STORE_SIZE, header, length);
   bzero(buff + LSN_STORE_SIZE + length,
         share->block_size - LSN_STORE_SIZE - KEYPAGE_CHECKSUM_SIZE - length);

storage/maria/ma_loghandler.c

@@ -1330,6 +1330,17 @@ static my_bool translog_close_log_file(TRANSLOG_FILE *file)
 }
 
 
+/**
+  @brief Dummy function for write failure (the log to not use
+  pagecache writing)
+*/
+
+void translog_dummy_write_failure(uchar *data __attribute__((unused)))
+{
+  return;
+}
+
+
 /**
   @brief Initializes TRANSLOG_FILE structure
 
@@ -1342,7 +1353,8 @@ static void translog_file_init(TRANSLOG_FILE *file, uint32 number,
                                my_bool is_sync)
 {
   pagecache_file_init(file->handler, &translog_page_validator,
-                      &translog_dummy_callback, file);
+                      &translog_dummy_callback,
+                      &translog_dummy_write_failure, file);
   file->number= number;
   file->was_recovered= 0;
   file->is_sync= is_sync;
@@ -2399,9 +2411,9 @@ static my_bool translog_recover_page_up_to_sector(uchar *page, uint16 offset)
 */
 
 static my_bool
-translog_dummy_callback(__attribute__((unused)) uchar *page,
-                        __attribute__((unused)) pgcache_page_no_t page_no,
-                        __attribute__((unused)) uchar* data_ptr)
+translog_dummy_callback(uchar *page __attribute__((unused)),
+                        pgcache_page_no_t page_no __attribute__((unused)),
+                        uchar* data_ptr __attribute__((unused)))
 {
   return 0;
 }

storage/maria/ma_open.c

@@ -1538,6 +1538,7 @@ static void set_data_pagecache_callbacks(PAGECACHE_FILE *file,
                                          MARIA_SHARE *share)
 {
   file->callback_data= (uchar*) share;
+  file->write_fail= &maria_page_write_failure;
   if (share->temporary)
   {
     file->read_callback=  &maria_page_crc_check_none;

storage/maria/ma_pagecache.c

@@ -613,7 +613,10 @@ static uint pagecache_fwrite(PAGECACHE *pagecache,
     lsn= lsn_korr(buffer + PAGE_LSN_OFFSET);
     DBUG_ASSERT(LSN_VALID(lsn));
     if (translog_flush(lsn))
+    {
+      (*filedesc->write_fail)(filedesc->callback_data);
       DBUG_RETURN(1);
+    }
   }
   DBUG_PRINT("info", ("write_callback: 0x%lx  data: 0x%lx",
                       (ulong) filedesc->write_callback,
@@ -624,8 +627,13 @@ static uint pagecache_fwrite(PAGECACHE *pagecache,
     DBUG_RETURN(1);
   }
 
-  DBUG_RETURN(my_pwrite(filedesc->file, buffer, pagecache->block_size,
-                        (pageno)<<(pagecache->shift), flags));
+  if (my_pwrite(filedesc->file, buffer, pagecache->block_size,
+                        (pageno)<<(pagecache->shift), flags))
+  {
+    (*filedesc->write_fail)(filedesc->callback_data);
+    DBUG_RETURN(1);
+  }
+  DBUG_RETURN(0);
 }
 
 

storage/maria/ma_pagecache.h

@@ -85,6 +85,7 @@ typedef struct st_pagecache_file
                            uchar *data);
   my_bool (*write_callback)(uchar *page, pgcache_page_no_t offset,
                             uchar *data);
+  void (*write_fail)(uchar *data);
   uchar *callback_data;
 } PAGECACHE_FILE;
 
@@ -257,9 +258,10 @@ extern void pagecache_unpin_by_link(PAGECACHE *pagecache,
 /* PCFLUSH_ERROR and PCFLUSH_PINNED. */
 #define PCFLUSH_PINNED_AND_ERROR (PCFLUSH_ERROR|PCFLUSH_PINNED)
 
-#define pagecache_file_init(F,RC,WC,D) \
+#define pagecache_file_init(F,RC,WC,WF,D) \
   do{ \
     (F).read_callback= (RC); (F).write_callback= (WC); \
+    (F).write_fail= (WF); \
     (F).callback_data= (uchar*)(D); \
   } while(0)
 

storage/maria/ma_pagecrc.c

@@ -290,3 +290,13 @@ my_bool maria_page_filler_set_none(uchar *page __attribute__((unused)),
 {
   return 0;
 }
+
+/**
+  @brief Write failure callback (mark table as corrupted)
+
+  @param data_ptr        Write callback data pointer (pointer to MARIA_SHARE)
+*/
+void maria_page_write_failure (uchar* data_ptr)
+{
+  maria_mark_crashed_share((MARIA_SHARE *)data_ptr);
+}

storage/maria/ma_panic.c

@@ -110,7 +110,8 @@ int maria_panic(enum ha_panic_function flag)
           pagecache_file_init(info->s->kfile, &maria_page_crc_check_index,
                               (info->s->options & HA_OPTION_PAGE_CHECKSUM ?
                                &maria_page_crc_set_index :
-                               &maria_page_filler_set_normal), info->s);
+                               &maria_page_filler_set_normal),
+                              &maria_page_write_failure, info->s);
         }
 	if (info->dfile.file < 0)
 	{
@@ -122,7 +123,8 @@ int maria_panic(enum ha_panic_function flag)
           pagecache_file_init(info->dfile, &maria_page_crc_check_data,
                               (share->options & HA_OPTION_PAGE_CHECKSUM ?
                                &maria_page_crc_set_normal:
-                               &maria_page_filler_set_normal), share);
+                               &maria_page_filler_set_normal),
+                              &maria_page_write_failure, share);
 	  info->rec_cache.file= info->dfile.file;
 	}
       }

storage/maria/maria_chk.c

@@ -1673,7 +1673,8 @@ static int maria_sort_records(HA_CHECK *param,
   pagecache_file_init(info->dfile, &maria_page_crc_check_data,
                       (share->options & HA_OPTION_PAGE_CHECKSUM ?
                        &maria_page_crc_set_normal :
-                       &maria_page_filler_set_normal), share);
+                       &maria_page_filler_set_normal),
+                      &maria_page_write_failure, share);
   info->state->del=0;
   info->state->empty=0;
   share->state.dellink= HA_OFFSET_ERROR;

storage/maria/maria_def.h

@@ -1090,4 +1090,6 @@ extern my_bool maria_page_filler_set_normal(uchar *page,
 extern my_bool maria_page_filler_set_none(uchar *page,
                                           pgcache_page_no_t page_no,
                                           uchar *data_ptr);
+extern void maria_page_write_failure(uchar* data_ptr);
+
 extern PAGECACHE *maria_log_pagecache;

storage/maria/unittest/ma_pagecache_consist.c

@@ -57,19 +57,31 @@ static uint flush_divider= 1000;
 #endif /*TEST_READERS*/
 #endif /*TEST_HIGH_CONCURENCY*/
 
+
 /**
   @brief Dummy pagecache callback.
 */
 
 static my_bool
-dummy_callback(__attribute__((unused)) uchar *page,
-               __attribute__((unused)) pgcache_page_no_t page_no,
-               __attribute__((unused)) uchar* data_ptr)
+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;
+}
+
+
 /*
   Get pseudo-random length of the field in (0;limit)
 
@@ -333,7 +345,8 @@ int main(int argc __attribute__((unused)),
 	    errno);
     exit(1);
   }
-  pagecache_file_init(file1, &dummy_callback, &dummy_callback, NULL);
+  pagecache_file_init(file1, &dummy_callback, &dummy_callback,
+                      &dummy_fail_callback, NULL);
   DBUG_PRINT("info", ("file1: %d", file1.file));
   if (chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO) != 0)
   {

storage/maria/unittest/ma_pagecache_single.c

@@ -60,19 +60,30 @@ static struct file_desc  simple_delete_flush_test_file[]=
   { 0, 0}
 };
 
+
 /**
   @brief Dummy pagecache callback.
 */
 
 static my_bool
-dummy_callback(__attribute__((unused)) uchar *page,
-               __attribute__((unused)) pgcache_page_no_t page_no,
-               __attribute__((unused)) uchar* data_ptr)
+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;
+}
+
 
 /*
   Recreate and reopen a file for test
@@ -520,7 +531,8 @@ int main(int argc __attribute__((unused)),
 	    errno);
     exit(1);
   }
-  pagecache_file_init(file1, &dummy_callback, &dummy_callback, NULL);
+  pagecache_file_init(file1, &dummy_callback, &dummy_callback,
+                      &dummy_fail_callback, NULL);
   my_close(tmp_file, MYF(0));
   my_delete(file2_name, MYF(0));
 

storage/maria/unittest/ma_test_loghandler_pagecache-t.c

@@ -20,19 +20,31 @@ static char *first_translog_file= (char*)"maria_log.00000001";
 static char *file1_name= (char*)"page_cache_test_file_1";
 static PAGECACHE_FILE file1;
 
+
 /**
   @brief Dummy pagecache callback.
 */
 
 static my_bool
-dummy_callback(__attribute__((unused)) uchar *page,
-               __attribute__((unused)) pgcache_page_no_t page_no,
-               __attribute__((unused)) uchar* data_ptr)
+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;
+}
+
+
 int main(int argc __attribute__((unused)), char *argv[])
 {
   uint pagen;
@@ -124,7 +136,8 @@ int main(int argc __attribute__((unused)), char *argv[])
 	    errno);
     exit(1);
   }
-  pagecache_file_init(file1, &dummy_callback, &dummy_callback, NULL);
+  pagecache_file_init(file1, &dummy_callback, &dummy_callback,
+                      &dummy_fail_callback, NULL);
   if (chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO) != 0)
   {
     fprintf(stderr, "Got error during file1 chmod() (errno: %d)\n",