Support parallelize option

2 files changed, 123 insertions, 19 deletions

diff --git a/README b/README
index e976856..4ff0254 100644
--- a/README
+++ b/README
@@ -28,3 +28,10 @@ option to mkzftree.
 mkzftree also accepts a -u option (uncompress), which can be used to
 convert a compressed tree back to normal form.  This can be used to
 read a zisofs CD-ROM on a machine without zisofs kernel support.
+
+This version of mkzftree supports a -p option (parallelize.)
+Specifying -p and a parallelism (e.g. -p4) allows up to four files to
+compress in parallel.  Depending on your setup and your data set size,
+this might speed things up if you are not completely I/O bound.  Use
+-p0 to completely disable parallel execution; this is the default.
+
diff --git a/mkzftree.c b/mkzftree.c
index 47d5489..928bea9 100644
--- a/mkzftree.c
+++ b/mkzftree.c
@@ -66,8 +66,10 @@
 #include <unistd.h>
 #include <limits.h>
 #include <utime.h>
+#include <signal.h>
 #include <sys/stat.h>
 #include <sys/types.h>
+#include <sys/wait.h>
 
 #include <zlib.h>
 
@@ -76,11 +78,76 @@
 /* Command line options */
 int force = 0;			/* Always compress */
 int level = 9;			/* Compression level */
+int parallel = 0;		/* Parallelism (0 = strictly serial) */
 int verbosity = 0;
 
 /* Program name */
 const char *program;
 
+/* Functions related to parallel execution */
+static volatile int work_threads = 0;
+static int is_worker = 0;
+
+/* This waits for one worker to finish */
+void wait_for_one_worker(void)
+{
+  int status;
+
+  if ( wait(&status) > 0 ) {
+    work_threads--;
+    
+    if ( WIFSIGNALED(status) || WEXITSTATUS(status) )
+      kill(getpid(), SIGTERM);	/* We had problems, stop now */
+  }
+}    
+
+/* This waits for *all* workers to finish */
+void wait_for_all_workers(void)
+{
+  while ( work_threads )
+    wait_for_one_worker();
+}
+
+/* This returns 1 if the "job" at hand should be performed */
+int spawn_worker(void)
+{
+  pid_t f;
+  
+  if ( parallel == 0 )
+    return 1;
+
+  fflush(NULL);
+
+  /* Wait for a work slot */
+  while ( work_threads >= parallel )
+    wait_for_one_worker();
+  
+  /* Spawn worker process */
+  work_threads++;		/* Avoids race conditions */
+  f = fork();
+  if ( f == -1 ) {
+    work_threads--;
+    return 1;			/* Do it ourselves */
+  }
+
+  if ( f == 0 ) {
+    /* Worker process */
+    is_worker = 1;
+    return 1;
+  } else {
+    /* Control process */
+    return 0;
+  }
+}
+
+/* Routine to perform at the end of the job */
+void end_worker(int err)
+{
+  if ( is_worker ) {
+    exit(err);
+  }
+}
+
 /* Convenience functions */
 void *xmalloc(size_t size)
 {
@@ -454,10 +521,11 @@ int block_compress_file(FILE *input, FILE *output, unsigned long size)
   return -1;
 }
 
-int munge_path(const char *inpath, const char *outpath, unsigned long size, munger_func munger)
+int munge_path(const char *inpath, const char *outpath, struct stat *st, munger_func munger)
 {
   FILE *in, *out;
-  int err, rv;
+  int err = 0, rv = 0;
+  struct utimbuf ut;
 
   in = fopen(inpath, "rb");
   if ( !in )
@@ -469,11 +537,31 @@ int munge_path(const char *inpath, const char *outpath, unsigned long size, mung
     errno = err;
     return -1;
   }
-  rv = munger(in, out, size);
 
-  err = errno;			/* Just in case */
-  fclose(in);
-  fclose(out);
+  if ( spawn_worker() ) {
+    rv = munger(in, out, st->st_size);
+    
+    err = rv ? errno : 0;
+    
+#ifdef HAVE_LCHOWN
+    lchown(outpath, st->st_uid, st->st_gid);
+#endif
+    if ( !S_ISLNK(st->st_mode) ) {
+#ifndef HAVE_LCHOWN
+      chown(outpath, st->st_uid, st->st_gid);
+#endif
+      chmod(outpath, st->st_mode);
+      ut.actime  = st->st_atime;
+      ut.modtime = st->st_mtime;
+      utime(outpath, &ut);
+    }
+
+    end_worker(err);
+  } else {
+    fclose(in);
+    fclose(out);
+  }
+
   errno = err;
   return rv;
 }
@@ -594,7 +682,7 @@ int munge_tree(const char *intree, const char *outtree, munger_func munger)
 	  }
 	} else {
 	  /* First encounter, compress and enter into hash */
-	  if ( munge_path(in_path, out_path, st.st_size, munger) ) {
+	  if ( munge_path(in_path, out_path, &st, munger) ) {
 	    fprintf(stderr, "%s: %s: %s", program, in_path, strerror(errno));
 	    err = 1;
 	    break;
@@ -603,7 +691,7 @@ int munge_tree(const char *intree, const char *outtree, munger_func munger)
 	}
       } else {
 	/* Singleton file; no funnies */
-	if ( munge_path(in_path, out_path, st.st_size, munger) ) {
+	if ( munge_path(in_path, out_path, &st, munger) ) {
 	  fprintf(stderr, "%s: %s: %s", program, in_path, strerror(errno));
 	  err = 1;
 	  break;
@@ -663,31 +751,35 @@ int munge_tree(const char *intree, const char *outtree, munger_func munger)
 	  }
       }
     }
+
+    /* This is done by munge_path() for files */
+    if ( !S_ISREG(st.st_mode) ) {
 #ifdef HAVE_LCHOWN
-    lchown(out_path, st.st_uid, st.st_gid);
+      lchown(out_path, st.st_uid, st.st_gid);
 #endif
-    if ( !S_ISLNK(st.st_mode) ) {
+      if ( !S_ISLNK(st.st_mode) ) {
 #ifndef HAVE_LCHOWN
-      chown(out_path, st.st_uid, st.st_gid);
+	chown(out_path, st.st_uid, st.st_gid);
 #endif
-      chmod(out_path, st.st_mode);
-      ut.actime  = st.st_atime;
-      ut.modtime = st.st_mtime;
-      utime(out_path, &ut);
+	chmod(out_path, st.st_mode);
+	ut.actime  = st.st_atime;
+	ut.modtime = st.st_mtime;
+	utime(out_path, &ut);
+      }
     }
   }
   closedir(thisdir);
-
+  
   free(in_path);
   free(out_path);
-
+  
   return err;
 }
 
 static void usage(int err)
 {
   fprintf(stderr,
-	  "Usage: %s [-vfhu] [-z level] intree outtree\n",
+	  "Usage: %s [-vfhu] [-p parallellism] [-z level] intree outtree\n",
 	  program);
   exit(err);
 }
@@ -702,7 +794,7 @@ int main(int argc, char *argv[])
 
   program = argv[0];
 
-  while ( (opt = getopt(argc, argv, "vfz:hu")) != EOF ) {
+  while ( (opt = getopt(argc, argv, "vfz:p:hu")) != EOF ) {
     switch(opt) {
     case 'f':
       force = 1;		/* Always compress */
@@ -725,6 +817,9 @@ int main(int argc, char *argv[])
     case 'u':
       munger = block_uncompress_file;
       break;
+    case 'p':
+      parallel = atoi(optarg);
+      break;
     default:
       usage(1);
       break;
@@ -750,6 +845,8 @@ int main(int argc, char *argv[])
 
   err = munge_tree(in, out, munger);
 
+  wait_for_all_workers();
+
   if ( err )
     exit(err);