add wait_all_children_with_timeout method

lib/Parallel/Prefork.pm

@@ -10,6 +10,8 @@ use List::Util qw/first max min/;
 use Proc::Wait3 ();
 use Time::HiRes ();
 
+use constant DEFAULT_WAIT_INTERVAL => 0.01;
+
 use Class::Accessor::Lite (
     rw => [ qw/max_workers spawn_interval err_respawn_interval trap_signals signal_received manager_pid on_child_reap before_fork after_fork/ ],
 );
@@ -200,6 +202,30 @@ sub wait_all_children {
     }
 }
 
+sub wait_all_children_with_timeout {
+    my ($self, $timeout, $interval) = @_;
+    return $self->wait_all_children() if !defined $timeout || $timeout <= 0;
+    $interval ||= DEFAULT_WAIT_INTERVAL;
+
+    $self->{_no_adjust_until} = undef;
+
+    my $start_at = [Time::HiRes::gettimeofday];
+    while (Time::HiRes::tv_interval($start_at) < $timeout) {
+        if (my ($pid) = $self->_wait(0)) {
+            if (delete $self->{worker_pids}{$pid}) {
+                $self->_on_child_reap($pid, $?);
+            }
+        }
+        last unless $self->num_workers;
+    }
+    continue {
+        Time::HiRes::sleep $interval;
+    }
+
+    return $self->num_workers;
+}
+
+
 sub _update_spawn_delay {
     my ($self, $secs) = @_;
     $self->{_no_adjust_until} = $secs ? Time::HiRes::time() + $secs : 0;
@@ -330,6 +356,11 @@ Sends signal to all worker processes.  Only usable from manager process.
 
 Blocks until all worker processes exit.  Only usable from manager process.
 
+=head2 wait_all_children_with_timeout($timeout, $interval = 0.01)
+
+Blocks until all worker processes exit or after C<$timeout> seconds.  Only usable from manager process.
+It repeated wait at predetermined intervals internally.  To change the interval, specify interval by second argument.
+
 =head1 AUTHOR
 
 Kazuho Oku

t/06-wait-all-children-with-timeout.t

@@ -0,0 +1,59 @@
+#! /usr/bin/perl
+
+use strict;
+use warnings;
+
+use Fcntl qw/:flock/;
+use File::Temp qw/tempfile/;
+use Test::More tests => 4;
+
+use Parallel::Prefork;
+
+my $reaped = 0;
+my $pm = Parallel::Prefork->new({
+    max_workers   => 3,
+    fork_delay    => 0,
+    on_child_reap => sub {
+        $reaped++;
+    }
+});
+
+my ($fh, $filename) = tempfile;
+syswrite $fh, '0', 1;
+close $fh;
+
+my $ppid = $$;
+until ($pm->signal_received) {
+    $pm->start and next;
+
+    open my $fh, '+<', $filename
+        or die "failed to open temporary file: $filename: ";
+    flock $fh, LOCK_EX;
+    sysread $fh, my $c, 10;
+    $c++;
+    seek $fh, 0, 0;
+    syswrite $fh, $c, length($c);
+    flock $fh, LOCK_UN;
+    close $fh;
+
+    my $rcv = 0;
+    local $SIG{TERM} = sub { $rcv++ };
+
+    if ($c == $pm->max_workers) {
+        kill 'TERM', $ppid;
+    }
+
+    1 while $rcv < $c;
+
+    $pm->finish;
+}
+is $pm->wait_all_children_with_timeout(1), 2, 'should reap one worker.';
+$pm->signal_all_children('TERM');
+is $pm->wait_all_children_with_timeout(1), 1, 'should reap one worker.';
+$pm->signal_all_children('TERM');
+$pm->wait_all_children();
+is $pm->num_workers, 0, 'all workers reaped.';
+
+is($reaped, $pm->max_workers, "properly called on_child_reap callback");
+
+unlink $filename;