Improve data throughput in dump by allowing the slaves to compress in parallel

diff --git a/CHANGES b/CHANGES
index 0000bb170288a39b609aa419535efb1567f41202..8796462fb08af9e7c663a508cee53735401c6127 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -1,4 +1,11 @@
-$Id: CHANGES,v 1.327 2011/06/10 12:17:23 stelian Exp $
+$Id: CHANGES,v 1.328 2011/06/10 12:41:53 stelian Exp $
+
+Changes between versions 0.4b44 and 0.4b45 (released ?????????????)
+===================================================================
+
+1.     Improve data throughput when using compression in dump by
+       allowing multiple slaves to compress in parallel. Thanks
+       to Phillip Susi <psusi@cfl.rr.com> for the patch.

 
 Changes between versions 0.4b43 and 0.4b44 (released June 10, 2011)
 ===================================================================
 
 Changes between versions 0.4b43 and 0.4b44 (released June 10, 2011)
 ===================================================================

diff --git a/dump/tape.c b/dump/tape.c
index e46d62630cc839884c481d2db406c51692a1db61..ec1b31117eb6e1092dcf68b43d576d3bb214bd88 100644 (file)
--- a/dump/tape.c
+++ b/dump/tape.c
@@ -37,7 +37,7 @@
 
 #ifndef lint
 static const char rcsid[] =
 
 #ifndef lint
 static const char rcsid[] =

-       "$Id: tape.c,v 1.93 2011/05/20 09:52:19 stelian Exp $";
+       "$Id: tape.c,v 1.94 2011/06/10 12:41:54 stelian Exp $";

 #endif /* not lint */
 
 #include <config.h>
 #endif /* not lint */
 
 #include <config.h>


@@ -178,10 +178,15 @@ static int tapea_volume;  /* value of spcl.c_tapea at volume start */
 
 int master;            /* pid of master, for sending error signals */
 int tenths;            /* length of tape overhead per block written */
 
 int master;            /* pid of master, for sending error signals */
 int tenths;            /* length of tape overhead per block written */

-static int caught;     /* have we caught the signal to proceed? */
-static int ready;      /* have we reached the lock point without having */
+static int caught1;    /* have we caught the signal to proceed? */
+static int ready1;     /* have we reached the lock point without having */

                        /* received the SIGUSR2 signal from the prev slave? */
                        /* received the SIGUSR2 signal from the prev slave? */

-static sigjmp_buf jmpbuf;      /* where to jump to if we are ready when the */
+static sigjmp_buf jmpbuf1;     /* where to jump to if we are ready when the */
+                       /* SIGUSR1 arrives from the previous slave */
+static int caught2;    /* have we caught the signal to proceed? */
+static int ready2;     /* have we reached the lock point without having */
+                       /* received the SIGUSR2 signal from the prev slave? */
+static sigjmp_buf jmpbuf2;     /* where to jump to if we are ready when the */

                        /* SIGUSR2 arrives from the previous slave */
 #ifdef USE_QFA
 static int gtperr = 0;
                        /* SIGUSR2 arrives from the previous slave */
 #ifdef USE_QFA
 static int gtperr = 0;


@@ -1028,15 +1033,26 @@ Exit(int status)
        exit(status);
 }
 
        exit(status);
 }
 

+/*
+ * proceed - handler for SIGUSR1, used to synchronize IO between the slaves.
+ */
+static void
+proceed1(UNUSED(int signo))
+{
+       if (ready1)
+               siglongjmp(jmpbuf1, 1);
+       caught1++;
+}
+

 /*
  * proceed - handler for SIGUSR2, used to synchronize IO between the slaves.
  */
 static void
 /*
  * proceed - handler for SIGUSR2, used to synchronize IO between the slaves.
  */
 static void

-proceed(UNUSED(int signo))
+proceed2(UNUSED(int signo))

 {
 {

-       if (ready)
-               siglongjmp(jmpbuf, 1);
-       caught++;
+       if (ready2)
+               siglongjmp(jmpbuf2, 1);
+       caught2++;

 }
 
 void
 }
 
 void


@@ -1058,16 +1074,21 @@ enslave(void)
        sigaction(SIGTERM, &sa, NULL); /* Slave sends SIGTERM on dumpabort() */
        sa.sa_handler = sigpipe;
        sigaction(SIGPIPE, &sa, NULL);
        sigaction(SIGTERM, &sa, NULL); /* Slave sends SIGTERM on dumpabort() */
        sa.sa_handler = sigpipe;
        sigaction(SIGPIPE, &sa, NULL);

-       sa.sa_handler = proceed;
+       sa.sa_handler = proceed1;
+       sa.sa_flags = SA_RESTART;
+       sigaction(SIGUSR1, &sa, NULL); /* Slave sends SIGUSR1 to next slave */
+       sa.sa_handler = proceed2;

        sa.sa_flags = SA_RESTART;
        sigaction(SIGUSR2, &sa, NULL); /* Slave sends SIGUSR2 to next slave */
    }
 
        for (i = 0; i < SLAVES; i++) {
                if (i == slp - &slaves[0]) {
        sa.sa_flags = SA_RESTART;
        sigaction(SIGUSR2, &sa, NULL); /* Slave sends SIGUSR2 to next slave */
    }
 
        for (i = 0; i < SLAVES; i++) {
                if (i == slp - &slaves[0]) {

-                       caught = 1;
+                       caught1 = 1;
+                       caught2 = 1;

                } else {
                } else {

-                       caught = 0;
+                       caught1 = 0;
+                       caught2 = 0;

                }
 
                if (socketpair(AF_UNIX, SOCK_STREAM, 0, cmd) < 0 ||
                }
 
                if (socketpair(AF_UNIX, SOCK_STREAM, 0, cmd) < 0 ||


@@ -1140,7 +1161,9 @@ killall(void)
  * previous process before writing to the tape, and sends SIGUSR2
  * to the next process when the tape write completes. On tape errors
  * a SIGUSR1 is sent to the master which then terminates all of the
  * previous process before writing to the tape, and sends SIGUSR2
  * to the next process when the tape write completes. On tape errors
  * a SIGUSR1 is sent to the master which then terminates all of the

- * slaves.
+ * slaves.  Each process sends SIGUSR1 to the next to signal that it
+ * is time to start reading from the disk, after it finishes reading
+ * and moves to the compression phase.

  */
 static void
 doslave(int cmd, 
  */
 static void
 doslave(int cmd, 


@@ -1177,6 +1200,7 @@ doslave(int cmd,
        sigset_t set;
 
        sigemptyset(&set);
        sigset_t set;
 
        sigemptyset(&set);

+       sigaddset(&set, SIGUSR1);

        sigaddset(&set, SIGUSR2);
        sigprocmask(SIG_BLOCK, &set, NULL);
        sigemptyset(&set);
        sigaddset(&set, SIGUSR2);
        sigprocmask(SIG_BLOCK, &set, NULL);
        sigemptyset(&set);


@@ -1240,6 +1264,15 @@ doslave(int cmd,
        while ((nread = dump_atomic_read( cmd, (char *)slp->req, reqsiz)) == reqsiz) {
                struct req *p = slp->req;
 
        while ((nread = dump_atomic_read( cmd, (char *)slp->req, reqsiz)) == reqsiz) {
                struct req *p = slp->req;
 

+               /* wait for previous slave to finish reading */
+               if (sigsetjmp(jmpbuf1, 1) == 0) {
+                       ready1 = 1;
+                       if (!caught1)
+                               sigsuspend(&set);
+               }
+               ready1 = 0;
+               caught1 = 0;
+

                for (trecno = 0; trecno < ntrec;
                     trecno += p->count, p += p->count) {
                        if (p->dblk) {  /* read a disk block */
                for (trecno = 0; trecno < ntrec;
                     trecno += p->count, p += p->count) {
                        if (p->dblk) {  /* read a disk block */


@@ -1252,6 +1285,8 @@ doslave(int cmd,
                                       quit("master/slave protocol botched.\n");
                        }
                }
                                       quit("master/slave protocol botched.\n");
                        }
                }

+               /* signal next slave to start reading */
+               (void) kill(nextslave, SIGUSR1);

 
                /* Try to write the data... */
                wrote = 0;
 
                /* Try to write the data... */
                wrote = 0;


@@ -1345,13 +1380,13 @@ doslave(int cmd,
                do_compress = compressed;
 #endif /* HAVE_ZLIB  || HAVE_BZLIB || HAVE_LZO */
 
                do_compress = compressed;
 #endif /* HAVE_ZLIB  || HAVE_BZLIB || HAVE_LZO */
 

-               if (sigsetjmp(jmpbuf, 1) == 0) {
-                       ready = 1;
-                       if (!caught)
+               if (sigsetjmp(jmpbuf2, 1) == 0) {
+                       ready2 = 1;
+                       if (!caught2)

                                sigsuspend(&set);
                }
                                sigsuspend(&set);
                }

-               ready = 0;
-               caught = 0;
+               ready2 = 0;
+               caught2 = 0;

 
 #ifdef USE_QFA
                if (gTapeposfd >= 0) {
 
 #ifdef USE_QFA
                if (gTapeposfd >= 0) {