* Incorporate code from on Anthony Towns's mrs, to allow running

  multiple jobs in parallel. The -j flag controls this.
  This can produce enormous speedups. For me, mr update takes
  12 minutes, while mr -j 10 update takes 1 minute!

diff --git a/TODO b/TODO
index 9eb4e900c1a44ffc9baf11893879a84f05cc6bb1..511ed8bf2f6e7768e26cad94533728b3fa77e8a8 100644 (file)
--- a/TODO
+++ b/TODO
@@ -11,8 +11,3 @@
 
   Until this is fixed, checkouts and updates need to be manually repeated
   after mrconfig files have changes.
-
-* Ability to run commands in paralell? (-j n)
-
-  If done right, this could make an update of a lot of repos faster. If
-  done wrong, it could suck mightily. ;-)

diff --git a/debian/changelog b/debian/changelog
index acdb9ee395368e6daad16125d4b07d7a2d0ee65f..e61ec88b300d0fd8890fdaf6434651f68e06b082 100644 (file)
--- a/debian/changelog
+++ b/debian/changelog
@@ -13,7 +13,13 @@ mr (0.5) UNRELEASED; urgency=low
   [ Alexander Wirt ]
   * Add support for mercurial.
 
- -- Joey Hess <joeyh@debian.org>  Sat, 20 Oct 2007 13:24:15 -0400
+  [ Joey Hess ]
+  * Incorporate code from on Anthony Towns's mrs, to allow running
+    multiple jobs in parallel. The -j flag controls this.
+    This can produce enormous speedups. For me, mr update takes
+    12 minutes, while mr -j 10 update takes 1 minute!
+
+ -- Joey Hess <joeyh@debian.org>  Sat, 20 Oct 2007 15:17:05 -0400
 
 mr (0.4) unstable; urgency=low
 

diff --git a/mr b/mr
index 65b0b1d3b4c5e3495590616d113b3a81a8091f4d..2b1a36c22b21df656ca592cca16439fd029ad00d 100755 (executable)
--- a/mr
+++ b/mr
@@ -157,6 +157,14 @@ about exactly which repositories failed and were skipped, if any.
 Just operate on the repository for the current directory, do not 
 recurse into deeper repositories.
 
+=item -j number
+
+Run the specified number of jobs in parallel. This can greatly speed up
+operations such as updates.
+
+Note that in -j mode, all output of the jobs goes to standard output, even
+output that would normally go to standard error.
+
 =back
 
 =head1 FILES
@@ -255,6 +263,13 @@ use warnings;
 use strict;
 use Getopt::Long;
 use Cwd qw(getcwd abs_path);
+use POSIX "WNOHANG";
+use constant {
+       OK => 0,
+       FAILED => 1,
+       SKIPPED => 2,
+       ABORT => 3,
+};
 
 $SIG{INT}=sub {
        print STDERR "mr: interrupted\n";
@@ -267,6 +282,7 @@ my $directory=getcwd();
 my $verbose=0;
 my $stats=0;
 my $no_recurse=0;
+my $jobs=1;
 my %config;
 my %configfiles;
 my %knownactions;
@@ -279,6 +295,7 @@ my $result=GetOptions(
        "v|verbose" => \$verbose,
        "s|stats" => \$stats,
        "n|no-recurse" => \$no_recurse,
+       "j|jobs=i" => \$jobs,
 );
 if (! $result || @ARGV < 1) {
        die("Usage: mr [-d directory] action [params ...]\n".
@@ -292,17 +309,17 @@ if (! $result || @ARGV < 1) {
 if ($ENV{MR_CONFIG} !~ /^\//) {
        $ENV{MR_CONFIG}=getcwd()."/".$ENV{MR_CONFIG};
 }
+# Try to set MR_PATH to the path to the program.
+eval {
+       use FindBin qw($Bin $Script);
+       $ENV{MR_PATH}=$Bin."/".$Script;
+};
 
 loadconfig(\*DATA);
 loadconfig($ENV{MR_CONFIG});
 #use Data::Dumper;
 #print Dumper(\%config);
 
-eval {
-       use FindBin qw($Bin $Script);
-       $ENV{MR_PATH}=$Bin."/".$Script;
-};
-
 # alias expansion and command stemming
 my $action=shift @ARGV;
 if (exists $alias{$action}) {
@@ -418,10 +435,39 @@ if (! @repos) {
        $nochdir=1;
 }
 
-my (@failed, @ok, @skipped);
-foreach my $repo (@repos) {
-       action($action, @$repo);
+# run the action on each repository and print stats
+my (@ok, @failed, @skipped);
+if ($jobs > 1) {
+       mrs(@repos);
+}
+else {
+       foreach my $repo (@repos) {
+               record($repo, action($action, @$repo));
+       }
+}
+if (! @ok && ! @failed && ! @skipped) {
+       die "mr $action: no repositories found to work on\n";
+}
+print "mr $action: finished (".join("; ",
+       showstat($#ok+1, "ok", "ok"),
+       showstat($#failed+1, "failed", "failed"),
+       showstat($#skipped+1, "skipped", "skipped"),
+).")\n";
+if ($stats) {
+       if (@skipped) {
+               print "mr $action: (skipped: ".join(" ", @skipped).")\n";
+       }
+       if (@failed) {
+               print STDERR "mr $action: (failed: ".join(" ", @failed).")\n";
+       }
+}
+if (@failed) {
+       exit 1;
 }
+elsif (! @ok && @skipped) {
+       exit 1;
+}
+exit 0;
 
 sub action { #{{{
        my ($action, $dir, $topdir, $subdir) = @_;
@@ -433,15 +479,14 @@ sub action { #{{{
        if ($action eq 'checkout') {
                if (-d $dir) {
                        print "mr $action: $dir already exists, skipping checkout\n" if $verbose;
-                       push @skipped, $dir;
-                       return;
+                       return SKIPPED;
                }
 
                $dir=~s/^(.*)\/[^\/]+\/?$/$1/;
 
                if (! -d $dir) {
                        print "mr $action: creating parent directory $dir\n" if $verbose;
-                       my $ret=system("mkdir", "-p", $dir);
+                       system("mkdir", "-p", $dir);
                }
        }
        elsif ($action eq 'update') {
@@ -460,27 +505,26 @@ sub action { #{{{
                if ($ret != 0) {
                        if (($? & 127) == 2) {
                                print STDERR "mr $action: interrupted\n";
-                               exit 2;
+                               return ABORT;
                        }
                        elsif ($? & 127) {
                                print STDERR "mr $action: skip test received signal ".($? & 127)."\n";
-                               exit 1;
+                               return ABORT;
                        }
                }
                if ($ret >> 8 == 0) {
                        print "mr $action: $dir skipped per config file\n" if $verbose;
-                       push @skipped, $dir;
-                       return;
+                       return SKIPPED;
                }
        }
        
        if (! $nochdir && ! chdir($dir)) {
                print STDERR "mr $action: failed to chdir to $dir: $!\n";
-               push @failed, $dir;
+               return FAILED;
        }
        elsif (! exists $config{$topdir}{$subdir}{$action}) {
                print STDERR "mr $action: no defined $action command for $topdir$subdir, skipping\n";
-               push @skipped, $dir;
+               return SKIPPED;
        }
        else {
                if (! $nochdir) {
@@ -494,34 +538,112 @@ sub action { #{{{
                        join(" ", map { s/\//\/\//g; s/"/\"/g; '"'.$_.'"' } @ARGV);
                print STDERR "mr $action: running >>$command<<\n" if $verbose;
                my $ret=system($command);
+               print "\n";
                if ($ret != 0) {
                        if (($? & 127) == 2) {
                                print STDERR "mr $action: interrupted\n";
-                               exit 2;
+                               return ABORT;
                        }
                        elsif ($? & 127) {
                                print STDERR "mr $action: received signal ".($? & 127)."\n";
+                               return ABORT;
                        }
                        print STDERR "mr $action: failed ($ret)\n" if $verbose;
-                       push @failed, $dir;
                        if ($ret >> 8 != 0) {
                                print STDERR "mr $action: command failed\n";
                        }
                        elsif ($ret != 0) {
                                print STDERR "mr $action: command died ($ret)\n";
                        }
+                       return FAILED;
                }
                else {
                        if ($action eq 'checkout' && ! -d $dir) {
                                print STDERR "mr $action: $dir missing after checkout\n";;
-                               push @failed, $dir;
-                               return;
+                               return FAILED;
                        }
 
-                       push @ok, $dir;
+                       return OK;
                }
+       }
+} #}}}
 
-               print "\n";
+# run actions on multiple repos, in parallel
+my %jobs;
+sub mrs { #{{{
+       $| = 1;
+       my @fhs;
+       my @out;
+       my $running=0;
+       while (@fhs or @repos) {
+               while ($running < $jobs && @repos) {
+                       $SIG{CHLD}='DEFAULT';
+                       $running++;
+                       my $repo = shift @repos;
+                       my $pid = open(my $fh, "-|");
+                       if (! $pid) {
+                               open(STDERR, ">&STDOUT");
+                               exit action($action, @$repo);
+                       }
+                       $jobs{$pid}=$repo;
+                       push @fhs, $fh;
+                       push @out, "";
+                       reaper();
+                       $SIG{CHLD}=\&reaper;
+               }
+               my ($rin, $rout) = ('','', '');
+               my $nfound;
+               foreach my $x (@fhs) {
+                       next unless defined $x;
+                       vec($rin, fileno($x), 1) = 1;
+               }
+               $nfound = select($rout=$rin, undef, undef, 1);
+               foreach my $i (0..$#fhs) {
+                       my $fh = $fhs[$i];
+                       next unless defined $fh;
+                       if (vec($rout, fileno($fh), 1) == 1) {
+                               my $r = '';
+                               if (sysread($fh, $r, 1024) == 0) {
+                                       close($fh);
+                                       $fhs[$i] = undef;
+                                       $running--;
+                                       print $out[$i];
+                                       $out[$i]='';
+                               }
+                               $out[$i] .= $r;
+                       }
+               }
+               while (@fhs and !defined $fhs[0]) {
+                       shift @fhs;
+                       shift @out;
+               }
+       }
+} #}}}
+
+sub reaper { #{{{
+       while ((my $pid = waitpid(-1, &WNOHANG)) > 0) {
+               record($jobs{$pid}, $? >> 8) if exists $jobs{$pid};
+       }
+} #}}}
+               
+sub record { #{{{
+       my $dir=shift()->[0];
+       my $ret=shift;
+
+       if ($ret == OK) {
+               push @ok, $dir;
+       }
+       elsif ($ret == FAILED) {
+               push @failed, $dir;
+       }
+       elsif ($ret == SKIPPED) {
+               push @skipped, $dir;
+       }
+       elsif ($ret == ABORT) {
+               exit 1;
+       }
+       else {
+               die "unknown exit status $ret";
        }
 } #}}}
 
@@ -534,29 +656,6 @@ sub showstat { #{{{
        }
        return;
 } #}}}
-if (! @ok && ! @failed && ! @skipped) {
-       die "mr $action: no repositories found to work on\n";
-}
-print "mr $action: finished (".join("; ",
-       showstat($#ok+1, "ok", "ok"),
-       showstat($#failed+1, "failed", "failed"),
-       showstat($#skipped+1, "skipped", "skipped"),
-).")\n";
-if ($stats) {
-       if (@skipped) {
-               print "mr $action: (skipped: ".join(" ", @skipped).")\n";
-       }
-       if (@failed) {
-               print STDERR "mr $action: (failed: ".join(" ", @failed).")\n";
-       }
-}
-if (@failed) {
-       exit 1;
-}
-elsif (! @ok && @skipped) {
-       exit 1;
-}
-exit 0;
 
 my %loaded;
 sub loadconfig { #{{{