Beginnings of a synchronicity model...to-tytso-20011015c

5 files changed, 79 insertions, 18 deletions

diff --git a/arena.c b/arena.c
index c8f0fa5..c95c1ce 100644
--- a/arena.c
+++ b/arena.c
@@ -34,6 +34,7 @@
 #include <sched.h>
 #include <sys/time.h>
 #include <sys/resource.h>
+#include <sys/wait.h>
 
 #include "objstore.h"
 #include "internals.h"
@@ -419,20 +420,59 @@ void *objstore_init(const char *main_file, const char *log_file, size_t *arena_l
  * checkpoints will have to wait!
  *
  * Set gc_factor to 0.0 to force a gc, and to HUGE_VAL to inhibit gc.
+ *
+ * The "wait" argument determines the level of synchronicity required.
+ *
+ * Returns 0 on synchronous completion, 1 if the checkpoint was skipped,
+ * (pid_t)-1 on error, or the pid_t of the asynchronous process.  Note:
+ * if an asynchronous process is launched, and the next call uses
+ * OBJSYNC_NONE, then it is up to the parent process to wait for the
+ * launched processes; this process only remembers the last process launched.
+ * FIXME: perhaps we should double-fork these processes?
  */
-int objstore_checkpoint(double gc_factor)
+pid_t objstore_checkpoint(double gc_factor, enum objsync_enum wait)
 {
+  static pid_t last_sync = 0;
   struct ObjStore *os = objstore_os_struct;
-  int f;
+  pid_t f, w;
   char *pi, *epi;
   void *page;
   
   pi  = os->pageinfo;
   epi = os->pageinfo + (os->arena_len >> os->pageshift);
 
+  if ( last_sync ) {
+    int status;
+    switch (wait) {
+    case OBJSYNC_NONE:
+      break;
+    case OBJSYNC_SKIP:
+      w = waitpid(last_sync, &status, WNOHANG);
+      if ( w == 0 )
+	return 1;		/* Skip */
+      if ( w == (pid_t)-1 ) {
+	/* Note: under strace, waitpid() will return -1 instead of 0 */
+	return (pid_t)-1;
+      } else if ( WIFSIGNALED(status) || (WIFEXITED(status) && WEXITSTATUS(status)) ) {
+	return (pid_t)-1;	/* Badness */
+      }
+      break;
+    case OBJSYNC_WAIT:
+    case OBJSYNC_SYNC:
+      w = waitpid(last_sync, &status, 0);
+      if ( w == (pid_t)-1 ) {
+	if ( errno != ECHILD )
+	  return (pid_t)-1;	/* Badness */
+      } else if ( WIFSIGNALED(status) || (WIFEXITED(status) && WEXITSTATUS(status)) ) {
+	return (pid_t)-1;	/* Badness */
+      }
+      break;
+    }
+  }
+
   f = fork();
-  if ( f < 0 )
-    return 1;			/* Checkpoint failed! */
+  if ( f == (pid_t)-1 )
+    return (pid_t)-1;		/* Checkpoint failed! */
   else if ( f > 0 ) {
     /* Parent process -- just mark all dirty pages clean */
     
@@ -466,7 +506,19 @@ int objstore_checkpoint(double gc_factor)
     }
     os->dirty_count = 0;	/* No pages dirty */
     os->fork_seq++;		/* Increase next sequence number */
-    return 0;
+
+    if ( wait == OBJSYNC_SYNC ) {
+      int status;
+      waitpid(f, &status, 0);
+      last_sync = 0;
+      if ( WIFSIGNALED(status) || (WIFEXITED(status) && WEXITSTATUS(status)) )
+	return (pid_t)-1;	/* Something very bad happened... */
+      else
+	return 0;
+    } else {
+      last_sync = f;
+      return f;
+    }
   } else {
     /* Child process -- do the actual work of writing back dirty pages */
 
@@ -487,7 +539,9 @@ int objstore_checkpoint(double gc_factor)
 
       /* Make sure we were indeed next in turn */
       lseek(os->log_fd, -(off_t)sizeof(last_rec), SEEK_END);
-      if ( objstore_read(os->log_fd, &last_rec, sizeof(last_rec)) < sizeof(last_rec)) {
+      if ( objstore_read(os->log_fd, &last_rec, sizeof(last_rec)) < sizeof(last_rec) ||
+	   last_rec.magic != LOGRECORD_MAGIC ) {
+	/* Something bad happened... */
 	  kill(getppid(), SIGABRT); /* Kill main process */
 	  _exit(99);
       }
diff --git a/lpsm.h b/lpsm.h
index a6c08ac..65875d0 100644
--- a/lpsm.h
+++ b/lpsm.h
@@ -26,8 +26,16 @@
 
 /* Low-level core routines */
 
+/* When checkpointing, specify the level of synchronicity required */
+enum objsync_enum {
+  OBJSYNC_NONE,			/* No synchronicity */
+  OBJSYNC_SKIP,			/* Skip if previous writeback still running */
+  OBJSYNC_WAIT,			/* Wait if previous writeback still running */
+  OBJSYNC_SYNC			/* Wait until this writeback finishes */
+};
+
 void *objstore_init(const char *, const char *, size_t *);
-int objstore_checkpoint(double);
+int objstore_checkpoint(double, enum objsync_enum);
 int objstore_extend(size_t);
 
 /* Arena management routines */
diff --git a/testalloc.c b/testalloc.c
index 9a6d3ed..4fe5ee7 100644
--- a/testalloc.c
+++ b/testalloc.c
@@ -70,7 +70,7 @@ int main(int argc, char *argv[])
   printf("Arena:   initialized, %d root pointers at %p\n",
 	 ROOT_DATA_POINTERS, root_ptr);
 
-  objstore_checkpoint(0.0);
+  objstore_checkpoint(0.0, OBJSYNC_WAIT);
 
   for ( i = 0 ; i < COUNT ; i++ ) {
     slot = rand() % SLOTS;
@@ -111,12 +111,13 @@ int main(int argc, char *argv[])
 
     if ( (i & 255) == 0 ) {
       printf("Arena:   checkpointing...\n");
-      objstore_checkpoint(0.1);
+      objstore_checkpoint(0.1, OBJSYNC_SKIP);
     }
     printf("Arena:   %ld bytes officially occupied\n", occupied);
   }
-
-  objstore_checkpoint(0.0);
+  
+  fprintf(stderr, "ready...\n");
+  objstore_checkpoint(0.0, OBJSYNC_SYNC); /* Synchronous checkpoint */
 
   return 0;
 }
diff --git a/testbuddy.c b/testbuddy.c
index 4cf02a8..3b8ea5e 100644
--- a/testbuddy.c
+++ b/testbuddy.c
@@ -97,7 +97,7 @@ int main(int argc, char *argv[])
     return 1;
   }
 
-  objstore_checkpoint(0.0);
+  objstore_checkpoint(0.0, OBJSYNC_WAIT);
 
   for ( i = 0 ; i < 64 ; i++ ) {
     areas[i] = objstore_malloc(testsizes[i]);
@@ -105,7 +105,7 @@ int main(int argc, char *argv[])
 	   testsizes[i], testsizes[i], areas[i]);
   }
 
-  objstore_checkpoint(0.0);
+  objstore_checkpoint(0.0, OBJSYNC_SYNC);
 
   return 0;
 }
diff --git a/teststore.c b/teststore.c
index ba85141..532e7aa 100644
--- a/teststore.c
+++ b/teststore.c
@@ -30,18 +30,16 @@ int main(int argc, char *argv[])
   strcpy((char *)buf + 8192, "This is the third page!");
 
   printf("Read from third page: %s\n", (char *)buf + 8192);
-  objstore_checkpoint(0.5);
+  objstore_checkpoint(0.5, OBJSYNC_WAIT);
 
   strcpy((char *)buf + 8192, "This is also the third page!");
 
   printf("Read from third page: %s\n", (char *)buf + 8192);
-  objstore_checkpoint(0.0);
+  objstore_checkpoint(0.0, OBJSYNC_WAIT);
 
   objstore_extend(65536);
   strcpy((char *)buf + 32768, "This is the ninth page!");
-  objstore_checkpoint(0.0);
-
-  sleep(5);
+  objstore_checkpoint(0.0, OBJSYNC_SYNC);
 
   return 0;
 }