WL#3234 "Maria - control file manager"

- fixes to the control file module
- unit test for it
- renames of all Maria files I created to start with ma_


storage/maria/ma_checkpoint.c:
  Rename: storage/maria/checkpoint.c -> storage/maria/ma_checkpoint.c
storage/maria/ma_checkpoint.h:
  Rename: storage/maria/checkpoint.h -> storage/maria/ma_checkpoint.h
storage/maria/ma_least_recently_dirtied.c:
  Rename: storage/maria/least_recently_dirtied.c -> storage/maria/ma_least_recently_dirtied.c
storage/maria/ma_least_recently_dirtied.h:
  Rename: storage/maria/least_recently_dirtied.h -> storage/maria/ma_least_recently_dirtied.h
storage/maria/ma_recovery.c:
  Rename: storage/maria/recovery.c -> storage/maria/ma_recovery.c
storage/maria/ma_recovery.h:
  Rename: storage/maria/recovery.h -> storage/maria/ma_recovery.h
storage/maria/Makefile.am:
  control file module and its unit test program
storage/maria/ma_control_file.c:
  DBUG_ tags. Fix for gcc warnings.
  log_no -> logno (I felt "_no" sounded like a standalone "No" word).
  ma_ prefix for some functions.
  last_checkpoint_lsn_at_startup -> last_checkpoint_lsn (no need
  to make special vars for the values at startup). Same for last_logno.
  ma_control_file_write_and_force() now updates last_checkpoint_lsn
  and last_logno, the idea being that they belong to the module,
  others should not update them.
  And thus when the module shuts down, it zeroes those vars.
storage/maria/ma_control_file.h:
  importing structs from Sanja to get the control file module to compile;
  we'll remove that when Sanja pushes the log handler.
  CONTROL_FILE_IMPOSSIBLE_LOGNO is 0, not FFFFFFFF.
storage/maria/ma_control_file_test.c:
  Unit test program for the Maria control file module.
  Modelled after other ma_test* files in this directory (so, does
  not follow the unit test framework recently introduced with libtap;
  TODO as a task on all ma_test* programs).
  We test that writing to the control file works, and re-reading from it
  too, we check (by reading the file by ourselves) that its content
  on disk is correct, and check that a corrupted control file is detected.

storage/maria/Makefile.am

@@ -27,7 +27,7 @@ pkglib_LIBRARIES =	libmaria.a
 bin_PROGRAMS =		maria_chk maria_pack maria_ftdump
 maria_chk_DEPENDENCIES=	$(LIBRARIES)
 maria_pack_DEPENDENCIES=$(LIBRARIES)
-noinst_PROGRAMS =	ma_test1 ma_test2 ma_test3 ma_rt_test ma_sp_test
+noinst_PROGRAMS =	ma_test1 ma_test2 ma_test3 ma_rt_test ma_sp_test ma_control_file_test
 noinst_HEADERS =	maria_def.h ma_rt_index.h ma_rt_key.h ma_rt_mbr.h ma_sp_defs.h ma_fulltext.h ma_ftdefs.h ma_ft_test1.h ma_ft_eval.h
 ma_test1_DEPENDENCIES=	$(LIBRARIES)
 ma_test2_DEPENDENCIES=	$(LIBRARIES)
@@ -53,8 +53,9 @@ libmaria_a_SOURCES =	ma_init.c ma_open.c ma_extra.c ma_info.c ma_rkey.c \
 			ma_ft_update.c ma_ft_boolean_search.c \
 			ma_ft_nlq_search.c ft_maria.c ma_sort.c \
 			ma_rt_index.c ma_rt_key.c ma_rt_mbr.c ma_rt_split.c \
-			ma_sp_key.c
-CLEANFILES =		test?.MA? FT?.MA? isam.log ma_test_all ma_rt_test.MA? sp_test.MA?
+			ma_sp_key.c \
+			ma_control_file.c
+CLEANFILES =		test?.MA? FT?.MA? isam.log ma_test_all ma_rt_test.MA? sp_test.MA? maria_control
 DEFS =
 
 SUFFIXES = .sh

storage/maria/control_file.c → storage/maria/ma_control_file.c

@@ -5,7 +5,7 @@
 */
 
 #include "maria_def.h"
-
+#include "ma_control_file.h"
 
 /* Here is the implementation of this module */
 
@@ -17,16 +17,22 @@
 /* total size should be < sector size for atomic write operation */
 #define CONTROL_FILE_MAGIC_STRING "MACF"
 #define CONTROL_FILE_MAGIC_STRING_OFFSET 0
-#define CONTROL_FILE_MAGIC_STRING_SIZE sizeof(CONTROL_FILE_MAGIC_STRING)
+#define CONTROL_FILE_MAGIC_STRING_SIZE 4
 #define CONTROL_FILE_LSN_OFFSET (CONTROL_FILE_MAGIC_STRING_OFFSET + CONTROL_FILE_MAGIC_STRING_SIZE)
 #define CONTROL_FILE_LSN_SIZE (4+4)
 #define CONTROL_FILE_FILENO_OFFSET (CONTROL_FILE_LSN_OFFSET + CONTROL_FILE_LSN_SIZE)
 #define CONTROL_FILE_FILENO_SIZE 4
 #define CONTROL_FILE_MAX_SIZE (CONTROL_FILE_FILENO_OFFSET + CONTROL_FILE_FILENO_SIZE)
 
-
-LSN last_checkpoint_lsn_at_startup;
-uint32 last_logno_at_startup;
+/*
+  This module owns these two vars.
+  uint32 is always atomically updated, but LSN is 8 bytes, we will need
+  provisions to ensure that it's updated atomically in
+  ma_control_file_write_and_force(). Probably the log mutex could be
+  used. TODO.
+*/
+LSN last_checkpoint_lsn;
+uint32 last_logno;
 
 
 /*
@@ -35,7 +41,7 @@ uint32 last_logno_at_startup;
 */
 static int control_file_fd;
 
-static void lsn8store(char *buffer, LSN *lsn)
+static void lsn8store(char *buffer, const LSN *lsn)
 {
   int4store(buffer, lsn->file_no);
   int4store(buffer + CONTROL_FILE_FILENO_SIZE, lsn->rec_offset);
@@ -53,21 +59,23 @@ static LSN lsn8korr(char *buffer)
   Initialize control file subsystem
 
   SYNOPSIS
-    control_file_create_or_open()
+    ma_control_file_create_or_open()
 
-  Looks for the control file. If absent, it's a fresh start, create file.
-  If present, read it to find out last checkpoint's LSN and last log.
+  Looks for the control file. If absent, it's a fresh start, creates file.
+  If present, reads it to find out last checkpoint's LSN and last log, updates
+  the last_checkpoint_lsn and last_logno global variables.
   Called at engine's start.
 
   RETURN
     0 - OK
     1 - Error
 */
-int control_file_create_or_open()
+int ma_control_file_create_or_open()
 {
   char buffer[CONTROL_FILE_MAX_SIZE];
   char name[FN_REFLEN];
   MY_STAT stat_buff;
+  DBUG_ENTER("ma_control_file_create_or_open");
 
   /*
     If you change sizes in the #defines, you at least have to change the
@@ -79,12 +87,12 @@ int control_file_create_or_open()
 
   /* name is concatenation of Maria's home dir and "control" */
   if (fn_format(name, "control", maria_data_root, "", MYF(MY_WME)) == NullS)
-    return 1;
+    DBUG_RETURN(1);
 
   if ((control_file_fd= my_open(name,
                                 O_CREAT | O_BINARY | /*O_DIRECT |*/ O_RDWR,
                                 MYF(MY_WME))) < 0)
-    return 1;
+    DBUG_RETURN(1);
 
   /*
     TODO: from "man fsync" on Linux:
@@ -96,16 +104,14 @@ int control_file_create_or_open()
   */
 
   if (my_stat(name, &stat_buff, MYF(MY_WME)) == NULL)
-    return 1;
+    DBUG_RETURN(1);
 
-  if (stat_buff.st_size < CONTROL_FILE_MAX_SIZE)
+  if ((uint)stat_buff.st_size < CONTROL_FILE_MAX_SIZE)
   {
     /*
       File shorter than expected (either we just created it, or a previous run
       crashed between creation and first write); do first write.
-    */
-    char buffer[CONTROL_FILE_MAX_SIZE];
-    /*
+
       To be safer we should make sure that there are no logs or data/index
       files around (indeed it could be that the control file alone was deleted
       or not restored, and we should not go on with life at this point).
@@ -117,38 +123,41 @@ int control_file_create_or_open()
       directory of logs, finding the newest log, reading it to find last
       checkpoint... Slow but can save your db.
     */
-    last_checkpoint_lsn_at_startup.file_no= CONTROL_FILE_IMPOSSIBLE_FILENO;
-    last_checkpoint_lsn_at_startup.rec_offset= 0;
-    last_logno_at_startup= CONTROL_FILE_IMPOSSIBLE_FILENO;
+    LSN imposs_lsn= CONTROL_FILE_IMPOSSIBLE_LSN;
+    uint32 imposs_logno= CONTROL_FILE_IMPOSSIBLE_FILENO;
 
     /* init the file with these "undefined" values */
-    return control_file_write_and_force(last_checkpoint_lsn_at_startup,
-                                        last_logno_at_startup,
-                                        CONTROL_FILE_WRITE_ALL);
+    DBUG_RETURN(ma_control_file_write_and_force(&imposs_lsn, imposs_logno,
+                                             CONTROL_FILE_WRITE_ALL));
   }
   /* Already existing file, read it */
   if (my_read(control_file_fd, buffer, CONTROL_FILE_MAX_SIZE,
               MYF(MY_FNABP | MY_WME)))
-    return 1;
+    DBUG_RETURN(1);
   if (memcmp(buffer + CONTROL_FILE_MAGIC_STRING_OFFSET,
              CONTROL_FILE_MAGIC_STRING, CONTROL_FILE_MAGIC_STRING_SIZE))
-    return 1;
-  last_checkpoint_lsn_at_startup= lsn8korr(buffer + CONTROL_FILE_LSN_OFFSET);
-  last_logno_at_startup= uint4korr(buffer + CONTROL_FILE_FILENO_OFFSET);
-  return 0;
+  {
+    /*
+      TODO: what is the good way to report the error? Knowing that this
+      happens at startup, probably stderr.
+    */
+    DBUG_PRINT("error", ("bad magic string"));
+    DBUG_RETURN(1);
+  }
+  last_checkpoint_lsn= lsn8korr(buffer + CONTROL_FILE_LSN_OFFSET);
+  last_logno= uint4korr(buffer + CONTROL_FILE_FILENO_OFFSET);
+  DBUG_RETURN(0);
 }
 
 
-#define CONTROL_FILE_WRITE_ALL 0 /* write all 3 objects */
-#define CONTROL_FILE_WRITE_ONLY_LSN 1
-#define CONTROL_FILE_WRITE_ONLY_LOGNO 2
 /*
-  Write information durably to the control file.
+  Write information durably to the control file; stores this information into
+  the last_checkpoint_lsn and last_logno global variables.
 
   SYNOPSIS
-    control_file_write_and_force()
+    ma_control_file_write_and_force()
     checkpoint_lsn       LSN of last checkpoint
-    log_no               last log file number
+    logno                last log file number
     objs_to_write         what we should write
 
   Called when we have created a new log (after syncing this log's creation)
@@ -163,38 +172,44 @@ int control_file_create_or_open()
     1 - Error
 */
 
-int control_file_write_and_force(LSN *checkpoint_lsn, uint32 log_no,
+int ma_control_file_write_and_force(const LSN *checkpoint_lsn, uint32 logno,
                                  uint objs_to_write)
 {
   char buffer[CONTROL_FILE_MAX_SIZE];
   uint start, size;
+  DBUG_ENTER("ma_control_file_write_and_force");
+
   memcpy(buffer + CONTROL_FILE_MAGIC_STRING_OFFSET,
          CONTROL_FILE_MAGIC_STRING, CONTROL_FILE_MAGIC_STRING_SIZE);
   /* write checkpoint LSN */
   if (checkpoint_lsn)
     lsn8store(buffer + CONTROL_FILE_LSN_OFFSET, checkpoint_lsn);
   /* write logno */
-  int4store(buffer + CONTROL_FILE_FILENO_OFFSET, log_no);
+  int4store(buffer + CONTROL_FILE_FILENO_OFFSET, logno);
   if (objs_to_write == CONTROL_FILE_WRITE_ALL)
   {
     start= CONTROL_FILE_MAGIC_STRING_OFFSET;
     size= CONTROL_FILE_MAX_SIZE;
+    last_checkpoint_lsn= *checkpoint_lsn;
+    last_logno= logno;
   }
   else if (objs_to_write == CONTROL_FILE_WRITE_ONLY_LSN)
   {
     start= CONTROL_FILE_LSN_OFFSET;
     size= CONTROL_FILE_LSN_SIZE;
+    last_checkpoint_lsn= *checkpoint_lsn;
   }
   else if (objs_to_write == CONTROL_FILE_WRITE_ONLY_LOGNO)
   {
     start= CONTROL_FILE_FILENO_OFFSET;
     size= CONTROL_FILE_FILENO_SIZE;
+    last_logno= logno;
   }
   else /* incorrect value of objs_to_write */
     DBUG_ASSERT(0);
-  return (my_pwrite(control_file_fd, buffer + start, size,
-                    start, MYF(MY_FNABP |  MY_WME)) ||
-          my_sync(control_file_fd, MYF(MY_WME)));
+  DBUG_RETURN(my_pwrite(control_file_fd, buffer + start, size,
+                        start, MYF(MY_FNABP |  MY_WME)) ||
+              my_sync(control_file_fd, MYF(MY_WME)));
 }
 
 
@@ -202,10 +217,18 @@ int control_file_write_and_force(LSN *checkpoint_lsn, uint32 log_no,
   Free resources taken by control file subsystem
 
   SYNOPSIS
-    control_file_end()
+    ma_control_file_end()
 */
 
-void control_file_end()
+void ma_control_file_end()
 {
+  DBUG_ENTER("ma_control_file_end");
   my_close(control_file_fd, MYF(MY_WME));
+  /*
+    As this module owns these variables, closing the module forbids access to
+    them (just a safety):
+  */
+  last_checkpoint_lsn= CONTROL_FILE_IMPOSSIBLE_LSN;
+  last_logno= CONTROL_FILE_IMPOSSIBLE_FILENO;
+  DBUG_VOID_RETURN;
 }

storage/maria/control_file.h → storage/maria/ma_control_file.h

@@ -7,40 +7,69 @@
 #ifndef _control_file_h
 #define _control_file_h
 
-/* indicate absence of the log file number */
-#define CONTROL_FILE_IMPOSSIBLE_LOGNO 0xFFFFFFFF
+/*
+  Not everybody needs to call the control file that's why control_file.h is
+  not in maria_def.h. However, policy or habit may want to change this.
+*/
+
+#ifndef REMOVE_WHEN_SANJA_PUSHES_LOG_HANDLER
+/*
+  this is to get the control file to compile, until Sanja pushes the log
+  handler which will supersede those definitions.
+*/
+typedef struct st_lsn {
+  uint32 file_no;
+  uint32 rec_offset;
+} LSN;
+#define maria_data_root "."
+#endif
+
+/*
+  indicate absence of the log file number; first log is always number 1, 0 is
+  impossible.
+*/
+#define CONTROL_FILE_IMPOSSIBLE_FILENO 0
+/* logs always have a header */
+#define CONTROL_FILE_IMPOSSIBLE_LOG_OFFSET 0
+/*
+  indicate absence of LSN.
+*/
+#define CONTROL_FILE_IMPOSSIBLE_LSN ((LSN){CONTROL_FILE_IMPOSSIBLE_FILENO,CONTROL_FILE_IMPOSSIBLE_LOG_OFFSET})
 
 /* Here is the interface of this module */
 
 /*
   LSN of the last checkoint
-  (if last_checkpoint_lsn_at_startup.file_no == CONTROL_FILE_IMPOSSIBLE_LOGNO
+  (if last_checkpoint_lsn.file_no == CONTROL_FILE_IMPOSSIBLE_FILENO
   then there was never a checkpoint)
 */
-extern LSN last_checkpoint_lsn_at_startup;
+extern LSN last_checkpoint_lsn;
 /*
-  Last log number at startup time (if last_logno_at_startup ==
-  CONTROL_FILE_IMPOSSIBLE_LOGNO then there is no log file yet)
+  Last log number (if last_logno ==
+  CONTROL_FILE_IMPOSSIBLE_FILENO then there is no log file yet)
 */
-extern uint32 last_logno_at_startup;
+extern uint32 last_logno;
 
 /*
   Looks for the control file. If absent, it's a fresh start, create file.
   If present, read it to find out last checkpoint's LSN and last log.
   Called at engine's start.
 */
-int control_file_create_or_open();
+int ma_control_file_create_or_open();
 
 /*
   Write information durably to the control file.
   Called when we have created a new log (after syncing this log's creation)
   and when we have written a checkpoint (after syncing this log record).
 */
-int control_file_write_and_force(LSN *checkpoint_lsn, uint32 log_no,
+#define CONTROL_FILE_WRITE_ALL 0 /* write all 3 objects */
+#define CONTROL_FILE_WRITE_ONLY_LSN 1
+#define CONTROL_FILE_WRITE_ONLY_LOGNO 2
+int ma_control_file_write_and_force(const LSN *checkpoint_lsn, uint32 logno,
                                  uint objs_to_write);
 
 
 /* Free resources taken by control file subsystem */
-void control_file_end();
+void ma_control_file_end();
 
 #endif

storage/maria/ma_control_file_test.c

+/* Copyright (C) 2006 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+/* Unit test of the control file module of the Maria engine */
+
+/* TODO: make it fit the mytap framework */
+
+/*
+  Note that it is not possible to test the durability of the write (can't
+  pull the plug programmatically :)
+*/
+
+#include "maria.h"
+#include "ma_control_file.h"
+#include <my_getopt.h>
+
+char file_name[FN_REFLEN];
+int fd= -1;
+
+static void clean_files();
+static void run_test_normal();
+static void run_test_abnormal();
+static void usage();
+static void get_options(int argc, char *argv[]);
+
+int main(int argc,char *argv[])
+{
+  MY_INIT(argv[0]);
+
+  get_options(argc,argv);
+
+  clean_files();
+  run_test_normal();
+  run_test_abnormal();
+
+  exit(0); /* all ok, if some test failed, we will have aborted */
+}
+
+/*
+  Abort unless given expression is non-zero.
+
+  SYNOPSIS
+    DIE_UNLESS(expr)
+
+  DESCRIPTION
+    We can't use any kind of system assert as we need to
+    preserve tested invariants in release builds as well.
+
+  NOTE
+    This is infamous copy-paste from mysql_client_test.c;
+    we should instead put it in some include in one single place.
+*/
+
+#define DIE_UNLESS(expr) \
+        ((void) ((expr) ? 0 : (die(__FILE__, __LINE__, #expr), 0)))
+#define DIE_IF(expr) \
+        ((void) (!(expr) ? 0 : (die(__FILE__, __LINE__, #expr), 0)))
+#define DIE(expr) \
+        die(__FILE__, __LINE__, #expr)
+
+void die(const char *file, int line, const char *expr)
+{
+  fprintf(stderr, "%s:%d: check failed: '%s'\n", file, line, expr);
+  abort();
+}
+
+
+static void clean_files()
+{
+  DIE_IF(fn_format(file_name, "control", maria_data_root, "", MYF(MY_WME)) ==
+         NullS);
+  my_delete(file_name, MYF(0)); /* maybe file does not exist, ignore error */
+}
+
+
+static void run_test_normal()
+{
+  LSN checkpoint_lsn;
+  uint32 logno;
+  uint objs_to_write;
+  uint i;
+  char buffer[4];
+
+  /* TEST0: Instance starts from scratch (control file does not exist) */
+  DIE_UNLESS(ma_control_file_create_or_open() == 0);
+  /* Check that the module reports no information */
+  DIE_UNLESS(last_logno == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.file_no == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset == CONTROL_FILE_IMPOSSIBLE_LOG_OFFSET);
+
+  /* TEST1: Simulate creation of one log */
+
+  objs_to_write= CONTROL_FILE_WRITE_ONLY_LOGNO;
+  logno= 123;
+  DIE_UNLESS(ma_control_file_write_and_force(NULL, logno,
+                                          objs_to_write) == 0);
+  /* Check that last_logno was updated */
+  DIE_UNLESS(last_logno == logno);
+  /* Simulate shutdown */
+  ma_control_file_end();
+  /* Verify amnesia */
+  DIE_UNLESS(last_logno == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.file_no == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset == CONTROL_FILE_IMPOSSIBLE_LOG_OFFSET);
+  /* And restart */
+  DIE_UNLESS(ma_control_file_create_or_open() == 0);
+  DIE_UNLESS(last_logno == logno);
+
+  /* TEST2: Simulate creation of 5 logs */
+
+  objs_to_write= CONTROL_FILE_WRITE_ONLY_LOGNO;
+  logno= 100;
+  for (i= 0; i<5; i++)
+  {
+    logno*= 3;
+    DIE_UNLESS(ma_control_file_write_and_force(NULL, logno,
+                                            objs_to_write) == 0);
+  }
+  ma_control_file_end();
+  DIE_UNLESS(last_logno == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.file_no == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset == CONTROL_FILE_IMPOSSIBLE_LOG_OFFSET);
+  DIE_UNLESS(ma_control_file_create_or_open() == 0);
+  DIE_UNLESS(last_logno == logno);
+
+  /*
+    TEST3: Simulate one checkpoint, one log creation, two checkpoints, one
+    log creation.
+  */
+
+  objs_to_write= CONTROL_FILE_WRITE_ONLY_LSN;
+  checkpoint_lsn= (LSN){5, 10000};
+  logno= 10;
+  DIE_UNLESS(ma_control_file_write_and_force(&checkpoint_lsn, logno,
+                                          objs_to_write) == 0);
+  /* check that last_logno was not updated */
+  DIE_UNLESS(last_logno != logno);
+  /* Check that last_checkpoint_lsn was updated */
+  DIE_UNLESS(last_checkpoint_lsn.file_no == checkpoint_lsn.file_no);
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset == checkpoint_lsn.rec_offset);
+
+  objs_to_write= CONTROL_FILE_WRITE_ONLY_LOGNO;
+  checkpoint_lsn= (LSN){5, 20000};
+  logno= 17;
+  DIE_UNLESS(ma_control_file_write_and_force(&checkpoint_lsn, logno,
+                                          objs_to_write) == 0);
+  /* Check that checkpoint LSN was not updated */
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset != checkpoint_lsn.rec_offset);
+  objs_to_write= CONTROL_FILE_WRITE_ONLY_LSN;
+  checkpoint_lsn= (LSN){17, 20000};
+  DIE_UNLESS(ma_control_file_write_and_force(&checkpoint_lsn, logno,
+                                          objs_to_write) == 0);
+  objs_to_write= CONTROL_FILE_WRITE_ONLY_LSN;
+  checkpoint_lsn= (LSN){17, 45000};
+  DIE_UNLESS(ma_control_file_write_and_force(&checkpoint_lsn, logno,
+                                          objs_to_write) == 0);
+  objs_to_write= CONTROL_FILE_WRITE_ONLY_LOGNO;
+  logno= 19;
+  DIE_UNLESS(ma_control_file_write_and_force(&checkpoint_lsn, logno,
+                                          objs_to_write) == 0);
+
+  ma_control_file_end();
+  DIE_UNLESS(last_logno == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.file_no == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset == CONTROL_FILE_IMPOSSIBLE_LOG_OFFSET);
+  DIE_UNLESS(ma_control_file_create_or_open() == 0);
+  DIE_UNLESS(last_logno == logno);
+  DIE_UNLESS(last_checkpoint_lsn.file_no == checkpoint_lsn.file_no);
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset == checkpoint_lsn.rec_offset);
+
+  /*
+    TEST4: actually check by ourselves the content of the file.
+    Note that constants (offsets) are hard-coded here, precisely to prevent
+    someone from changing them in the control file module and breaking
+    backward-compatibility.
+  */
+
+  DIE_IF((fd= my_open(file_name,
+                      O_BINARY | O_RDWR,
+                      MYF(MY_WME))) < 0);
+  DIE_IF(my_read(fd, buffer, 16, MYF(MY_FNABP |  MY_WME)) != 0);
+  DIE_IF(my_close(fd, MYF(MY_WME)) != 0);  
+  i= uint4korr(buffer+4);
+  DIE_UNLESS(i == last_checkpoint_lsn.file_no);
+  i= uint4korr(buffer+8);
+  DIE_UNLESS(i == last_checkpoint_lsn.rec_offset);
+  i= uint4korr(buffer+12);
+  DIE_UNLESS(i == last_logno);
+
+
+  /* TEST5: Simulate stop/start/nothing/stop/start */
+
+  ma_control_file_end();
+  DIE_UNLESS(last_logno == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(ma_control_file_create_or_open() == 0);
+  ma_control_file_end();
+  DIE_UNLESS(last_logno == CONTROL_FILE_IMPOSSIBLE_FILENO);
+  DIE_UNLESS(ma_control_file_create_or_open() == 0);
+  DIE_UNLESS(last_logno == logno);
+  DIE_UNLESS(last_checkpoint_lsn.file_no == checkpoint_lsn.file_no);
+  DIE_UNLESS(last_checkpoint_lsn.rec_offset == checkpoint_lsn.rec_offset);
+
+}
+
+static void run_test_abnormal()
+{
+  /* Corrupt the control file */
+  DIE_IF((fd= my_open(file_name,
+                      O_BINARY | O_RDWR,
+                      MYF(MY_WME))) < 0);
+  DIE_IF(my_write(fd, "papa", 4, MYF(MY_FNABP |  MY_WME)) != 0);
+  DIE_IF(my_close(fd, MYF(MY_WME)) != 0);
+
+  /* Check that control file module sees the problem */
+  DIE_IF(ma_control_file_create_or_open() == 0);
+}
+
+
+static struct my_option my_long_options[] =
+{
+#ifndef DBUG_OFF
+  {"debug", '#', "Debug log.",
+   0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+#endif
+  {"help", '?', "Display help and exit",
+   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"version", 'V', "Print version number and exit",
+   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
+  { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
+};
+
+
+static void version()
+{
+  printf("ma_control_file_test: unit test for the control file "
+         "module of the Maria storage engine. Ver 1.0 \n");
+}
+
+static my_bool
+get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
+	       char *argument)
+{
+  switch(optid) {
+  case 'V':
+    version();
+    exit(0);
+  case '#':
+    DBUG_PUSH (argument);
+    break;
+  case '?':
+    version();
+    usage();
+    exit(0);
+  }
+  return 0;
+}
+
+
+/* Read options */
+
+static void get_options(int argc, char *argv[])
+{
+  int ho_error;
+
+  if ((ho_error=handle_options(&argc, &argv, my_long_options, get_one_option)))
+    exit(ho_error);
+
+  return;
+} /* get options */
+
+
+static void usage()
+{
+  printf("Usage: %s [options]\n\n", my_progname);
+  my_print_help(my_long_options);
+  my_print_variables(my_long_options);
+}