Improve Jobs package:

[flac-archive] / fa-flacd

diff --git a/fa-flacd b/fa-flacd
index 83be598de0b6564d950d6e520ae4079af569da46..20641358e068227c450e8cb94f7bbc500f6af7cc 100755 (executable)
--- a/fa-flacd
+++ b/fa-flacd
@@ -18,57 +18,96 @@ package Jobs;
 use strict;
 use warnings;
 
+use Errno;
 use POSIX ':sys_wait_h';
 
-our @jobs;
-our @finished;
-
-sub reaper {
-    while ((my $pid = waitpid(-1, WNOHANG)) > 0) {
-        push(@finished, [$pid, $?]);
-    }
-
-    # XXX if $pid == -1 handle errors?
-
-    $SIG{CHLD} = \&reaper;
-}
-
 sub newjob {
     my $f = shift;
-    my %o = @_;
+    my $jobs = shift;
+    my $debug = shift;
     my $pid;
 
-    $SIG{CHLD} = \&reaper;
-    if (not $o{'debug'}) {
+    if (not $debug) {
         $pid = fork();
         if (not defined($pid)) {
             die("fork: $!");
         }
     }
 
-    if ($o{'debug'} or $pid == 0) {
-        $SIG{CHLD} = 'DEFAULT';
+    if ($debug or $pid == 0) {
         exit($f->());
     }
 
-    push(@jobs, $pid);
+    if ($pid == 0) {
+        exit($f->());
+    }
+
+    push(@$jobs, $pid);
 
     return $pid;
 }
 
 sub deljob {
-    my $i = shift;
-    my $j;
+    my $pid = shift;
+    my $status = shift;
+    my $jobs = shift;
+
+    for (my $i = 0; $i <= $#$jobs; $i++) {
+        if ($pid == $jobs->[$i]) {
+            splice(@$jobs, $i, 1);
+            last;
+        }
+    }
+
+    return ($pid, $status);
+}
 
-    my ($pid, $status) = @{$finished[$i]};
+sub run {
+    my %o = @_;
+    my $maxjobs = $o{'max-jobs'};
+    my $get_job = $o{'get-job'};
+    my $notify_start = $o{'notify-start'};
+    my $notify_finish = $o{'notify-finish'};
+    my @jobs;
+    my $pid;
 
-    for ($j = 0; $j <= $#jobs; $j++) {
-        $pid == $jobs[$j] and splice(@jobs, $j, 1) and last;
+    # Call notifier function if given.
+    sub call {
+        my $f = shift or return;
+        ref($f) eq 'CODE' or return;
+        $f->(@_);
     }
 
-    splice(@finished, $i, 1);
+    while (1) {
+        if (@jobs <= $maxjobs) {
+            my $job;
+            while (defined($job = $get_job->())) {
+                $pid = newjob($job, \@jobs, $o{'debug'});
+                call($notify_start, $pid, @jobs);
+                @jobs < $maxjobs or last;
+            }
+
+            # No jobs running and get-job returned undef; we're finished.
+            if (@jobs == 0 and not defined($job)) {
+                return;
+            }
+        }
 
-    return ($pid, $status);
+        # Now running as many jobs as we can, block waiting for one to die.
+        do {
+            $pid = waitpid(-1, 0);
+        } while ($pid == 0
+                 or ($pid == -1 and ($!{ECHILD} or $!{EINTR})));
+        $pid == -1 and die("waitpid(-1): $!");
+
+        # Before starting more, see if any others have finished.
+        do {
+            call($notify_finish, deljob($pid, $?, \@jobs), @jobs);
+        } while (($pid = waitpid(-1, WNOHANG)) > 0);
+        if ($pid == -1) {
+            $!{ECHILD} or $!{EINTR} or die("waitpid(-1): $!");
+        }
+    }
 }
 
 \f
@@ -281,39 +320,46 @@ sub flac {
 
 sub flacloop {
     my $MAXJOBS = shift;
-    my $i;
-
+    my @jobs;
+    my $dir;
 
     while (1) {
-        if (scalar(@jobs) <= $MAXJOBS) {
-            foreach $i (glob('*/tags')) {
-                my $dir = dirname($i);
-                my $pid =
-                  Jobs::newjob(sub {
-                                   open(STDERR, ">$dir/log")
-                                     or die("open(STDERR, >$dir/log): $!");
-                                   return flac($dir);
-                               }, 'debug'=>$debug);
-                verbose("new job $pid for $dir\n");
-                @Jobs::jobs <= $MAXJOBS or last;
-            }
-        }
-
-        for ($i = 0; $i <= $#finished; $i++) {
-            my ($pid, $status) = Jobs::deljob($i);
-            verbose("$pid finished (");
-            if (WIFEXITED($status)) {
-                verbose('exited with status ', WEXITSTATUS($status));
-            } elsif (WIFSIGNALED($status)) {
-                verbose('killed with signal ', WTERMSIG($status));
-            } elsif (WIFSTOPPED($status)) {
-                verbose('stopped with signal ', WSTOPSIG($status));
-            }
-            verbose(")\n");
+        if (scalar(@jobs = glob('*/tags')) == 0) {
+            sleep(5);
+            next;
         }
 
-        verbose(scalar(@jobs), " jobs\n");
-        sleep(5);
+        Jobs::run('max-jobs'=>$MAXJOBS,
+                  'debug'=>$debug,
+                  'get-job'=>sub {
+                      my $bork = shift(@jobs) or return;
+                      $dir = dirname($bork);
+                      return sub {
+                          open(STDERR, ">$dir/log")
+                            or die("open(STDERR, >$dir/log): $!");
+                          return flac($dir);
+                      }
+                  },
+
+                  'notify-start'=>sub {
+                      my $pid = shift;
+                      verbose("new job $pid for $dir\n");
+                      verbose(scalar(@_), " jobs\n");
+                  },
+
+                  'notify-finish'=>sub {
+                      my $pid = shift;
+                      my $status = shift;
+                      verbose("$pid finished (");
+                      if (WIFEXITED($status)) {
+                          verbose('exited with status ', WEXITSTATUS($status));
+                      } elsif (WIFSIGNALED($status)) {
+                          verbose('killed with signal ', WTERMSIG($status));
+                      } elsif (WIFSTOPPED($status)) {
+                          verbose('stopped with signal ', WSTOPSIG($status));
+                      }
+                      verbose(")\n");
+                  });
     }
 }