- Clean up documentation.lpsm-0.1.8

- Add lpsm_recover().
- Add zalloc test.
- Be better about returning useful error information in errno.

10 files changed, 281 insertions, 266 deletions

diff --git a/CHANGES b/CHANGES
index 367bc32..156c32c 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,11 @@
+Changes in 0.1.8:
+-----------------
+* Add lpsm_recover()
+* Try to make sure errno contains useful information if we return
+  failure.
+* More documentation cleanup.  Remove most of the stuff in the README
+  file which now lives in man pages.
+
 Changes in 0.1.7:
 -----------------
 * Work around what seems to be a bug in gcc 2.91.66.
diff --git a/Makefile b/Makefile
index 9a8870a..8ab7e8d 100644
--- a/Makefile
+++ b/Makefile
@@ -11,9 +11,10 @@
 ##
 ## -----------------------------------------------------------------------
 
-TEST     = teststore test_mmap ftrunctest testbuddy testalloc testrecovery
+TEST     = teststore testbuddy testalloc testrecovery zallocevil \
+	   test_mmap ftrunctest 
 SONAME   = libpsm.so.0
-VERSION  = 0.1.7
+VERSION  = 0.1.8
 LIBPSM	 = libpsm.so.$(VERSION) libpsm.a
 
 CFILES   = arena.c bitops.c \
@@ -114,10 +115,13 @@ testalloc: testalloc.o libpsm.a
 testrecovery: testrecovery.o libpsm.a
 	$(CC) $(LDFLAGS) -o $@ $< $(LIBS)
 
+zallocevil: zallocevil.o libpsm.a
+	$(CC) $(LDFLAGS) -o $@ $< $(LIBS)
+
 test_mmap: test_mmap.o
 	$(CC) $(LDFLAGS) -o $@ $<
 
-ftrunctest: ftrunctest.o
+ftrunctest: ftrunctest.o 
 	$(CC) $(LDFLAGS) -o $@ $<
 
 .depend:
diff --git a/README b/README
index 4d2aa35..a4ec808 100644
--- a/README
+++ b/README
@@ -1,4 +1,15 @@
 This is a test release of the Linux Persistent Memory system.
+The LPSM is a C library with a simple interface that manages a segment
+of memory backed by a persistent file. LPSM differs from ordinary
+mmap() in two ways: it can optionally offer heap management
+(malloc()/free()/realloc()) within the arena; more importantly, a
+transaction log is used to ensure the consistency of the persistent
+representation. The application notifies LPSM whenever the arena is in
+a consistent state, suitable for "checkpointing". If the application
+or system crashes, the arena will always be recovered to a consistent
+checkpoint.
+
+
 
 Unless otherwise noted, all files in this distributions are:
 
@@ -51,188 +62,16 @@ used for the commit log.
 
 The LPSM library installs a signal handler for SIGSEGV.
 
+Please see the following man pages included with this distribution:
 
-* Functions to use in unmanaged mode:
-
-void *lpsm_arena_init(const char *datafile, const char *logfile,
-                      size_t *arena_size, void *arena_base)
-
-	Open the persistent memory arena that consists of the two
-	files datafile and logfile.  If these files don't exist, they
-	are created.  The argument "arena_size" contains the minimum
-	size of the arena (if smaller, it will be resized to this
-	size.)  On exit, it contains the actual size of the arena.
-
-	It returns a pointer to the arena, or NULL on error.
-
-	If arena_base is set, try to map the arena at that particular
-	base address.  If arena_base is NULL, use an
-	architecture-specific default value.
-
-
-pid_t lpsm_checkpoint(double ratio, enum psmsync wait)
-
-	Called to indicate that the memory arena is currently in a
-	consistent state and is safe to checkpoint.  On a crash, the
-	memory arena is ALWAYS guaranteed to be returned to the state
-	as of one of the calls to lpsm_checkpoint(); which exact one
-	depends on the "wait" argument.
-
-	The "ratio" parameter indicates if a log flush should take
-	place; if the logfile is at least "ratio" times the size of
-	the datafile, a log flush (replay the log into the datafile
-	and truncate the logfile) will happen once the log has been
-	written.  Specify 0.0 to do a log flush every time, HUGE_VAL
-	to specify that no log flush should happen.
-
-	The "wait" parameter controls the level of asynchronicity of
-	the checkpoint process:
-
-	PSMSYNC_NONE    - A checkpoint is unconditionally forked, and
-	not waited for.  The application is responsible for handing
-	or ignoring the resulting SIGCHLD.  This provides no
-	guarantees that a checkpoint is complete until the SIGCHLD
-	comes back.
-
-	PSMSYNC_SKIP	- wait() for the previous checkpoint process,
-	if it is still running, skip this checkpoint opportunity.
-	This provides no guarantees for which checkpoint opportunity
-	will be used in the event of a crash, but is useful if
-	checkpoint opportunities are very frequent, and continued
-	execution of the master process is important.
-
-	PSMSYNC_WAIT	- wait() for the previous checkpoint process,
-	if it is still running, sleep until the previous checkpoint
-	process has finished.  This guarantees that in the event of a
-	crash, the recovery point will be no later that one checkpoint
-	*before* the latest checkpoint executed.
-
-	PSMSYNC_SYNC	- wait() for the previous *and* current
-	checkpoint processes.  This guarantees that upon exit of
-	lpsm_checkpoint() the persistent memory is fully consistent on
-	disk.  This guarantees that in the event of a crash, the
-	recovery point will be no later than the latest checkpoint
-	executed.  This is typically used with a "ratio" argument of
-	0.0 to perform a clean shutdown.
-
-	lpsm_checkpoint() returns (pid_t)-1 on error; on success, it returns
-	0 if no asynchronous checkpoint process is pending, otherwise
-	the pid of the asynchronous checkpoint process.
-	
-
-int lpsm_extend(size_t newsize)
-
-	Extends the size of the arena to "newsize".  Returns 0 on
-	success, -1 on error.
-
-
-void lpsm_shutdown(void)
-
-	Shuts down the persistent memory system, frees all resources,
-	and restores the previous state of the SIGSEGV signal handler.
-	IT DOES NOT PERFORM A CHECKPOINT BEFORE DOING SO -- normally a
-	clean shutdown is performed as:
-
-		lpsm_checkpoint(0.0, PSMSYNC_SYNC);
-		lpsm_shutdown();
-
-	Calling lpsm_shutdown() followed by lpsm_arena_init() or
-	lpsm_init() can be used in exceptional events to reinitialize
-	the arena to the last checkpointed state.  This is very slow,
-	however.
-
-
-* Functions to use in managed mode:
-
-void **lpsm_init(const char *datafile, const char *logfile)
-
-	Opens/creates the managed persistent memory store defined by
-	datafile and logfile.  On success, returns a pointer to an
-	array of LPSM_ROOT_POINTERS (32) "root pointers" (void *),
-	which are available for the application to point to its
-	fundamental data structures.  On failure, returns NULL.
-
-
-void *lpsm_malloc(size_t size)
-
-	Allocates "size" bytes of persistent memory.  Returns a
-	pointer to the newly allocated memory, or NULL on failure.
-	The memory is uninitialized.
-
-
-void lpsm_free(void *ptr)
-
-	Frees the allocation pointed to by ptr.
-
-
-void *lpsm_realloc(void *ptr, size_t new_size)
-
-	Attempts to resize the allocation pointed to by ptr to size
-	"new_size".  Returns a pointer to the new allocation on
-	success, NULL on failure.  If the returned pointer is
-	different from ptr, the contents of the old allocation is
-	copied into the new allocation.  If new_size is greater than
-	the old size, the bytes beyond the old allocation size are
-	uninitialized.
-
-	If ptr == NULL, this call is equivalent to lpsm_malloc().
-
-	If new_size is zero, this call is equivalent to:
-
-	   lpsm_free(ptr);
-	   return NULL;
-
-void *lpsm_zalloc(size_t size)
-void *lpsm_calloc(size_t nmemb, size_t size)
-
-	Same as lpsm_malloc(), but returns memory that is initialized
-	to zero.  The implementation attempts to be smart, and for
-	large blocks can be significantly more efficient than
-	lpsm_malloc() followed by memset().
-
-	lpsm_calloc() is implemented as an inline or macro which
-	returns lpsm_zalloc(nmemb*size).
-
-
-struct lpsm_alloc_stats {
-  size_t size;
-  size_t free;
-  size_t alloc;
-};
-struct lpsm_alloc_stats *lpsm_alloc_stats(void);
-
-	Returns an array, malloc()'d in conventional (transient)
-	memory, containing allocation statistics: for each block size,
-	the size in bytes, number of blocks free, and number of blocks
-	allocated (in use).  The list terminates with an entry with
-	size == 0.  It is the responsibility of the application to
-	free() the allocated list.
-
-	Returns NULL on failure.
-
-
-pid_t lpsm_checkpoint(double ratio, enum psmsync wait)
-
-	Identical function as in unmanaged mode.
-
-
-void lpsm_shutdown(void)
-
-	Identical function as in unmanaged mode.
-
-
-The memory allocator used is a hybrid buddy system/slab allocator,
-which has the following properties:
+Unmanaged mode:
 
-* Allocation of large blocks are always page-aligned; allocation of
-  powers of 2 is very efficient;
+	lpsm_arena_init(3)
+	lpsm_recover(3)
 
-* Allocation of small blocks can be done efficiently down to single
-  bytes.  Allocation of small blocks (2032 bytes or less in the
-  current implementation) are aligned to a 16-byte datum unless they
-  are 8 bytes or smaller in size (in which case they are aligned to
-  the nearest higher power of 2.)
+Managed mode:
 
-* The allocator is very sensitive to bogus pointers passed to
-  lpsm_free() or lpsm_realloc().  Passing bogus pointers to these
-  routines is very likely to result in arena corruption or crash.
+	lpsm_init(3)
+	lpsm_malloc(3)
+	lpsm_arena_stats(3)
+	lpsm_recover(3)
diff --git a/arena.c b/arena.c
index ff8d2bf..6c5a1b0 100644
--- a/arena.c
+++ b/arena.c
@@ -174,27 +174,34 @@ static void lpsm_sigsegv(int signal, siginfo_t *siginfo, void *ptr)
  * as well as during-execution garbage collect.
  * THIS ROUTINE SHOULD BE INVOKED WITH LOCK HELD ON THE LOG FILE.
  */
-static int lpsm_log_writeback(void)
+static int lpsm_log_writeback(struct lpsm_arena *pm)
 {
   struct lpsm_logrecord record;
   off_t position, last_commit;
   struct flock lockmain;
   
   last_commit = 0;		/* Last COMMIT record found */
-  position = lseek(PM->log_fd, 0, SEEK_SET);
+  position = lseek(pm->log_fd, 0, SEEK_SET);
   
-  while ( lpsm_read(PM->log_fd, &record, sizeof(record)) == sizeof(record) ) {
+  while ( lpsm_read(pm->log_fd, &record, sizeof(record)) == sizeof(record) ) {
     if ( record.magic != LOGRECORD_MAGIC )
       break;			/* Bad magic, assume rest of log corrupt */
-    if ( record.record_type == osrec_commit ) {
+
+    switch ( record.record_type ) {
+    case osrec_commit:
       /* NOTE: last_commit points to the final byte to examine, thus
 	 at the *end* of the final commit record. */
       position += sizeof(record);
       last_commit = position;	/* Found a commit record */
-    } else if ( record.record_type == osrec_page ) {
+      break;
+
+    case osrec_page:
       /* Advance past current page cluster */
-      position = lseek(PM->log_fd, record.size, SEEK_CUR);
-    } else {
+      position = lseek(pm->log_fd, record.size, SEEK_CUR);
+      break;
+
+    default:
+      errno = EINVAL;
       return -1;		/* Unknown record - unsafe to process */
     }
   }
@@ -202,55 +209,64 @@ static int lpsm_log_writeback(void)
   /* Now we know where the last commit was.  Now we can process
      everything up to that point. */
   
-  position = lseek(PM->log_fd, 0, SEEK_SET);
+  position = lseek(pm->log_fd, 0, SEEK_SET);
   
-  while ( lpsm_read(PM->log_fd, &record, sizeof(record))
+  while ( lpsm_read(pm->log_fd, &record, sizeof(record))
 	  == sizeof(record) && position < last_commit ) {
     if ( record.magic != LOGRECORD_MAGIC )
       break;			/* Bad magic, assume rest of log corrupt */
-    if ( record.record_type == osrec_commit ) {
+
+    switch ( record.record_type ) {
+    case osrec_commit:
       /* Found a commit record, do nothing */
       position += sizeof(record);
-    } else if ( record.record_type == osrec_page ) {
-      /* Write back data to file */
-      char *data;
-      
-      position += sizeof(record);
-      
-      lockmain.l_type   = F_WRLCK;
-      lockmain.l_whence = SEEK_SET;
-      lockmain.l_start  = record.offset;
-      lockmain.l_len    = record.size;
-      while ( fcntl(PM->main_fd, F_SETLKW, &lockmain) == -1 && errno == EINTR ); 
-      data = mmap(NULL, record.size, PROT_WRITE, MAP_SHARED,
-		  PM->main_fd, record.offset);
-      if ( data == MAP_FAILED )
-	return -1;
-      if ( lpsm_read(PM->log_fd, data, record.size) != record.size )
-	return -1;		/* Badness */
-      if ( munmap(data, record.size) )
-	return -1;
-      
-      lockmain.l_type = F_UNLCK;
-      while ( fcntl(PM->main_fd, F_SETLKW, &lockmain) == -1 && errno == EINTR );
-      position += record.size;
-    } else {
+      break;
+
+    case osrec_page:
+      {
+	/* Write back data to file */
+	char *data;
+	
+	position += sizeof(record);
+	
+	lockmain.l_type   = F_WRLCK;
+	lockmain.l_whence = SEEK_SET;
+	lockmain.l_start  = record.offset;
+	lockmain.l_len    = record.size;
+	while ( fcntl(pm->main_fd, F_SETLKW, &lockmain) == -1 && errno == EINTR ); 
+	data = mmap(NULL, record.size, PROT_WRITE, MAP_SHARED,
+		    pm->main_fd, record.offset);
+	if ( data == MAP_FAILED )
+	  return -1;
+	if ( lpsm_read(pm->log_fd, data, record.size) != record.size )
+	  return -1;		/* Badness */
+	if ( munmap(data, record.size) )
+	  return -1;
+	
+	lockmain.l_type = F_UNLCK;
+	while ( fcntl(pm->main_fd, F_SETLKW, &lockmain) == -1 && errno == EINTR );
+	position += record.size;
+      }
+      break;
+
+    default:
+      errno = EINVAL;
       return -1;		/* Unknown record - unsafe to process */
     }
   }
   
   /* Log successfully recovered.  Truncate. */
-  fsync(PM->main_fd);
-  ftruncate(PM->log_fd, 0);
+  fsync(pm->main_fd);
+  ftruncate(pm->log_fd, 0);
   /* Write initial commit record, for sequence number recovery */
   record.magic       = LOGRECORD_MAGIC;
   record.record_type = osrec_commit;
-  record.size        = PM->fork_seq;
+  record.size        = pm->fork_seq;
   record.offset      = 0x54494d43; /* For debugging */
-  if ( lpsm_write(PM->log_fd, &record, sizeof(record)) < sizeof(record) )
+  if ( lpsm_write(pm->log_fd, &record, sizeof(record)) < sizeof(record) )
     return -1;
 
-  fsync(PM->log_fd);		/* Indicate log recovery complete */
+  fsync(pm->log_fd);		/* Indicate log recovery complete */
 
   return 0;
 }
@@ -258,31 +274,81 @@ static int lpsm_log_writeback(void)
 /*
  * Routine to do log recovery
  */
-static int lpsm_recover_log(void)
+static int lpsm_recover_log(struct lpsm_arena *pm)
 {
   struct flock lock;
   int rv = 0;
+  int rerrno;
 
   /* First, lock the log file */
   lock.l_type   = F_WRLCK;
   lock.l_whence = SEEK_SET;
   lock.l_start  = 0;
   lock.l_len    = 0;
-  while ( fcntl(PM->log_fd, F_SETLKW, &lock) == -1 && errno == EINTR );
+  while ( fcntl(pm->log_fd, F_SETLKW, &lock) == -1 && errno == EINTR );
 
   /* Do log recovery, and write initial commit record. */
-  rv = lpsm_log_writeback();
+  rv = lpsm_log_writeback(pm);
+  rerrno = errno;
 
   /* Increase the sequence number, since we just wrote a commit. */
-  PM->fork_seq++;
+  pm->fork_seq++;
 
   /* Unlock file and run. */
   lock.l_type = F_UNLCK;
-  while ( fcntl(PM->log_fd, F_SETLKW, &lock) == -1 && errno == EINTR );
+  while ( fcntl(pm->log_fd, F_SETLKW, &lock) == -1 && errno == EINTR );
 
+  errno = rerrno;
   return rv;
 }
 
+static int
+lpsm_open_and_recover(struct lpsm_arena *pm, const char *main_file,
+		      const char *log_file)
+{
+  int myerrno;
+
+  pm->fork_seq = 0;		/* Initialize sequence counter */
+  pm->main_fd = pm->log_fd = 0;
+
+  pm->main_fd = open(main_file, O_RDWR|O_CREAT, 0666);
+  if ( pm->main_fd < 0 )
+    goto errx1;
+
+  pm->pagesize = getpagesize();
+  if ( pm->pagesize & (pm->pagesize - 1) )
+    goto errx2;			/* WTF -- pagesize not a power of 2? */
+
+  /* Compute log2(pm->pagesize) */
+  pm->pageshift = 0;
+  while ( (1 << pm->pageshift) < pm->pagesize )
+    pm->pageshift++;
+
+  /*
+   * Open log file
+   */
+  pm->log_fd = open(log_file, O_RDWR|O_APPEND|O_CREAT, 0666);
+  if ( pm->log_fd < 0 )
+    goto errx2;
+
+  /* Now, do log recovery if needed */
+  if ( lpsm_recover_log(pm) )
+    goto errx3;
+
+  return 0;
+
+ errx3:
+  myerrno = errno;
+  close(pm->log_fd);
+  errno = myerrno;
+ errx2:
+  myerrno = errno;
+  close(pm->main_fd);
+  errno = myerrno;
+ errx1:
+  return -1;
+}
+
 /*
  * Opens the object store.  This includes log
  * playback (crash recovery) if the log file exists
@@ -291,6 +357,7 @@ static int lpsm_recover_log(void)
 void *lpsm_arena_init(const char *main_file, const char *log_file,
 		      size_t *arena_len, void *arena_ptr)
 {
+  int myerrno;
   struct sigaction sigact;
   struct flock lock;
   off_t file_len, len = arena_len ? *arena_len : 0;
@@ -302,32 +369,9 @@ void *lpsm_arena_init(const char *main_file, const char *log_file,
   if ( !PM )
     goto errx0;
 
-  PM->fork_seq = 0;		/* Initialize sequence counter */
-
-  PM->main_fd = open(main_file, O_RDWR|O_CREAT, 0666);
-  if ( PM->main_fd < 0 )
+  if ( lpsm_open_and_recover(PM, main_file, log_file) )
     goto errx1;
 
-  PM->pagesize = getpagesize();
-  if ( PM->pagesize & (PM->pagesize - 1) )
-    goto errx2;			/* WTF -- pagesize not a power of 2? */
-
-  /* Compute log2(PM->pagesize) */
-  PM->pageshift = 0;
-  while ( (1 << PM->pageshift) < PM->pagesize )
-    PM->pageshift++;
-
-  /*
-   * Open log file
-   */
-  PM->log_fd = open(log_file, O_RDWR|O_APPEND|O_CREAT, 0666);
-  if ( PM->log_fd < 0 )
-    goto errx3;
-
-  /* Now, do log recovery if needed */
-  if ( lpsm_recover_log() )
-    goto errx3;
-
   /* Allocate arena memory space */
   lock.l_type = F_WRLCK;
   lock.l_whence = SEEK_SET;
@@ -373,21 +417,44 @@ void *lpsm_arena_init(const char *main_file, const char *log_file,
   return PM->arena;
 
  errx5:
+  myerrno = errno;
   munmap(PM->pageinfo, len >> PM->pageshift);
+  errno = myerrno;
  errx4:
+  myerrno = errno;
   munmap(arena_ptr, len);
+  errno = myerrno;
  errx3:
+  myerrno = errno;
   if ( PM->log_fd >= 0 ) close(PM->log_fd);
- errx2:
   close(PM->main_fd);
+  errno = myerrno;
  errx1:
+  myerrno = errno;
   free(PM);
+  errno = myerrno;
  errx0:
 
   return NULL;
 }
 
 /*
+ * Do an offline log recovery (exported function.)
+ */
+int lpsm_recover(const char *mainfile, const char *logfile)
+{
+  struct lpsm_arena pm;
+  int rv;
+
+  rv = lpsm_open_and_recover(&pm, mainfile, logfile);
+  if ( !rv ) {
+    close(pm.log_fd);
+    close(pm.main_fd);
+  }
+  return rv;
+}
+
+/*
  * Object store checkpoint.  Writes entries to the log file.
  * The "gc_factor" is the factor of maximum log size file relative
  * to the arena size.  For example, if gc_factor == 0.5 then if the
@@ -551,7 +618,7 @@ pid_t lpsm_checkpoint(double gc_factor, enum psmsync wait)
 	 them out of the shadow array.  The biggest problem with that
 	 is that it probably can't be done in the background, unlike
 	 this method.  Leave this as-is for now. */
-      if ( lpsm_log_writeback() ) {
+      if ( lpsm_log_writeback(PM) ) {
 	kill(getppid(), SIGABRT);
 	_exit(99);
       }
@@ -580,7 +647,7 @@ int lpsm_extend(size_t new_size)
   int ft;
   size_t add_size, old_size;
   size_t add_pages, old_pages, new_pages, file_pages;
-  struct rlimit fsizelimit;
+  int realerrno;
 
   old_size = PM->arena_len;
 
@@ -590,12 +657,6 @@ int lpsm_extend(size_t new_size)
   new_size = (new_size + PM->pagesize - 1) & ~(PM->pagesize - 1);
   add_size = new_size - old_size;
 
-  /* Make sure we won't run afoul of a set resource limit. */
-  if ( getrlimit(RLIMIT_FSIZE, &fsizelimit) == 0 &&
-       fsizelimit.rlim_cur != RLIM_INFINITY &&
-       fsizelimit.rlim_cur < new_size )
-    return -1;			/* Would exceed the file limit. */
-
   lock.l_type   = F_WRLCK;
   lock.l_whence = SEEK_SET;
   lock.l_start  = 0;
@@ -609,15 +670,19 @@ int lpsm_extend(size_t new_size)
   else
     ft = 0;
 
-  if ( ft )
+  if ( ft ) {
+    realerrno = errno;
     goto reset_size;		/* Failure */
+  }
 
   newp = mmap((char*)PM->arena + old_size,
 	      add_size, PROT_READ,
 	      MAP_PRIVATE|MAP_FIXED, PM->main_fd, old_size);
   
-  if ( newp == MAP_FAILED )
+  if ( newp == MAP_FAILED ) {
+    realerrno = errno;
     goto reset_size;		/* Failure */
+  }
 
   /* Since we specified MAP_FIXED, this should be guaranteed */
   assert( newp == (char*)PM->arena + old_size );
@@ -631,6 +696,7 @@ int lpsm_extend(size_t new_size)
 
   infop = realloc(PM->pageinfo, new_pages);
   if ( !infop ) {
+    realerrno = errno;
     munmap(newp, add_size);
     goto reset_size;
   }
@@ -654,6 +720,8 @@ int lpsm_extend(size_t new_size)
   ftruncate(PM->main_fd, file_size);
   /* Drop lock */
   while ( fcntl(PM->main_fd, F_SETLKW, &lock) == -1 && errno == EINTR );
+  /* Set errno to the real error */
+  errno = realerrno;
   return -1;
 }
 
diff --git a/lpsm.h b/lpsm.h
index 0af85bf..d458f18 100644
--- a/lpsm.h
+++ b/lpsm.h
@@ -38,11 +38,12 @@ enum psmsync {
   PSMSYNC_WAIT,			/* Wait if previous writeback still running */
   PSMSYNC_SYNC			/* Wait until this writeback finishes */
 };
-
+  
 void *lpsm_arena_init(const char *, const char *, size_t *, void *);
 pid_t lpsm_checkpoint(double, enum psmsync);
 int lpsm_extend(size_t);
 void lpsm_shutdown(void);
+int lpsm_recover(const char *, const char *);
 
 /* Arena management routines */
 
diff --git a/lpsm_arena_init.3.in b/lpsm_arena_init.3.in
index 1a91472..138dde4 100644
--- a/lpsm_arena_init.3.in
+++ b/lpsm_arena_init.3.in
@@ -169,8 +169,15 @@ returns 0 on success and -1 on failure.
 .PP
 .B lpsm_shutdown()
 does not return a value.
+.SH "ERRORS"
+On failure,
+.B lpsm_arena_init()
+will set
+.I errno
+to the appropriate value for the error that occurred.
 .SH "SEE ALSO"
 .BR lpsm_init (3),
+.BR lpsm_recover (3),
 .BR lpsm_malloc (3).
 .SH BUGS
 Checkpointing failures are currently not handled gracefully.
diff --git a/lpsm_init.3.in b/lpsm_init.3.in
index a91b21f..e07473a 100644
--- a/lpsm_init.3.in
+++ b/lpsm_init.3.in
@@ -149,9 +149,21 @@ on failure, or the pid of the spawned asychronous process.
 .PP
 .B lpsm_shutdown()
 does not return a value.
+.SH "ERRORS"
+On error,
+.B lpsm_init()
+will set
+.I errno
+to the appropriate error value.  In particular, if the files specified
+do not contain a valid managed-mode LPSM arena compatible with this
+machine architecture,
+.I errno
+is set to
+.BR EINVAL .
 .SH "SEE ALSO"
 .BR lpsm_malloc (3),
-.BR lpsm_arena_init (3).
+.BR lpsm_arena_init (3),
+.BR lpsm_recover (3).
 .SH BUGS
 Checkpointing failures are currently not handled gracefully.
 .PP
diff --git a/lpsm_recover.3.in b/lpsm_recover.3.in
new file mode 100644
index 0000000..8e6a9af
--- /dev/null
+++ b/lpsm_recover.3.in
@@ -0,0 +1,49 @@
+.\" -*- nroff -*- ---------------------------------------------------------
+.\"   
+.\"   Copyright 2001 H. Peter Anvin - All Rights Reserved
+.\"
+.\"   This program is free software; you can redistribute it and/or modify
+.\"   it under the terms of the GNU Lesser General Public License as
+.\"   published by the Free Software Foundation, Inc.,
+.\"   59 Temple Place Ste 330, Bostom MA 02111-1307, USA; version 2.1,
+.\"   incorporated herein by reference.
+.\"
+.\" -----------------------------------------------------------------------
+.\" $Id$
+.TH LPSM_RECOVER 3  "25 October 2001" "LPSM @@VERSION@@" "Linux Persistent Memory"
+.SH NAME
+lpsm_recovery \- Perform offline recovery of an LPSM database
+.SH SYNOPSIS
+.nf
+.B #include <lpsm.h>
+.sp
+.BI "int lpsm_recover(const char *" datafile ", const char *" logfile ");"
+.nl
+.fi
+.SH DESCRIPTION
+.B lpsm_recover()
+performs an offline log recovery of the specified pair of LPSM files.
+After log recovery is complete, the contents of
+.I datafile
+is identical to the contents of the LPSM arena, and
+.I logfile
+is truncated to its minimum size.
+.PP
+It is safe to call
+.B lpsm_recovery()
+while a different arena is mapped.
+.SH "RETURN VALUES"
+.B lpsm_recovery()
+returns 0 on success and -1 on failure; in the latter case
+.I errno
+is set to indicate the nature of the error.
+.SH "NOTES"
+It is not required to perform a log recovery before calling
+.B lpsm_init()
+or
+.BR lpsm_arena_init() .
+These routines perform log recovery as part of their standard startup
+sequence.
+.SH "SEE ALSO"
+.BR lpsm_init (3),
+.BR lpsm_arena_init (3).
diff --git a/mgmt.c b/mgmt.c
index 956520a..82ac80a 100644
--- a/mgmt.c
+++ b/mgmt.c
@@ -29,6 +29,7 @@
 #include <inttypes.h>
 #include <limits.h>
 #include <string.h>
+#include <errno.h>
 
 #include "lpsm.h"
 #include "internals.h"
@@ -127,6 +128,7 @@ void **lpsm_init(const char *main_file, const char *log_file)
        AH->arena_bytesize  != CHAR_BIT   ||
        AH->arena_ptrsize   != sizeof(void *) ||
        AH->buddy_order_min != BUDDY_ORDER_MIN ) {
+    errno = EINVAL;
     return NULL;		/* Incompatible or corrupt arena */
   }
 
@@ -136,8 +138,11 @@ void **lpsm_init(const char *main_file, const char *log_file)
     lpsm_shutdown();		/* Nothing written... */
     if ( !(AH = lpsm_arena_init(main_file, log_file, &arena_len, arena_base)) )
       return NULL;		/* Failed to initialize arena */
-    if ( AH->arena_base != (void *)AH )
+    if ( AH->arena_base != (void *)AH ) {
+      lpsm_shutdown();
+      errno = EINVAL;
       return NULL;		/* Something went weird... */
+    }
   }
 
   return AH->root_data_ptrs;
diff --git a/zallocevil.c b/zallocevil.c
new file mode 100644
index 0000000..ab666e8
--- /dev/null
+++ b/zallocevil.c
@@ -0,0 +1,22 @@
+/* Simple test program for zalloc() */
+
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+#include <limits.h>
+#include "lpsm.h"
+
+int main(int argc, char **argv) {
+   void **root_ptr,*ptr;
+   int i;
+
+   if (!(root_ptr = lpsm_init("arena.dat", "arena.log")))
+       return 1;
+   for(i=0;i<100000;i+=100) {
+       printf("%d\n",i);
+       ptr = lpsm_zalloc(i);
+       /*lpsm_free(ptr);*/
+   }
+   return 0;
+}