File Coverage

blib/lib/Parallel/Batch.pm

Criterion	Covered	Total	%
statement	45	50	90.0
branch	11	20	55.0
condition			n/a
subroutine	7	7	100.0
pod	1	1	100.0
total	64	78	82.0

package Parallel::Batch;

use 5.008;
use strict;
use warnings;

use POSIX qw/:sys_wait_h/;
use parent qw/Class::Accessor::Fast/;
Parallel::Batch->mk_accessors(qw/jobs code maxprocs progress_cb/);

our $VERSION = '0.04';


sub run {
    my $self = shift;

    my $procs = 0;
    my @jobs = @{$self->jobs};
    my %inprogress;
    # progress start(count)
    $self->_send_progress(start => scalar @jobs);

    while (@jobs) {
	if ($procs < $self->maxprocs) {
	    my $job = shift @jobs;
	    my $pid = fork();
	    die unless defined $pid;
	    if ($pid == 0) {
		$self->code->($job);
		exit;
	    }
	    if ($pid > 0) {
		$procs++;
                # progress new child
                $inprogress{$pid} = $job;
                $self->_send_progress('new', $job);
	    }
	}
	if ($procs == $self->maxprocs) {
	    if ((my $pid = wait()) > 0) {
		$procs--;
                # progress child finished
                $self->_send_progress('finish', delete $inprogress{$pid});
	    }
	}
	if ((my $pid = waitpid(-1, WNOHANG)) > 0) {
	    $procs--;
            $self->_send_progress('finish', delete $inprogress{$pid});
	}
    }
    while ($procs > 0) {
	if ((my $pid = wait()) > 0) {
	    $procs--;
            $self->_send_progress('finish', delete $inprogress{$pid});
	}
    }
    # progress done
    $self->_send_progress('done');
}

sub _send_progress {
    my ($self, $type, $arg) = @_;

    return unless $self->progress_cb;
    my $cb = $self->progress_cb->{$type};
    if (defined $cb)
    {
        $cb->($arg);
    }
}

1;
__END__