File Coverage

blib/lib/Parallel/Pipes/App.pm

Criterion	Covered	Total	%
statement	43	46	93.4
branch	15	24	62.5
condition			n/a
subroutine	5	5	100.0
pod	1	1	100.0
total	64	76	84.2

package Parallel::Pipes::App;
use strict;
use warnings;

use Parallel::Pipes;

our $VERSION = '0.101';

sub _min { $_[0] < $_[1] ? $_[0] : $_[1] }

sub run {
    my ($class, %argv) = @_;

    my $work = $argv{work} or die "need 'work' argument\n";
    my $num = $argv{num} or die "need 'num' argument\n";
    my $tasks = $argv{tasks} or die "need 'tasks' argument\n";

    my $before_work = $argv{before_work};
    my $after_work = $argv{after_work};

    my @result;
    my $pipes = Parallel::Pipes->new($num, $work);
    while (1) {
        my @ready = $pipes->is_ready;
        if (my @written = grep { $_->is_written } @ready) {
            for my $written (@written) {
                my $result = $written->read;
                if ($after_work) {
                    $after_work->($result);
                } else {
                    push @result, $result;
                }
            }
        }
        if (@$tasks) {
            my $min = _min $#{$tasks}, $#ready;
            for my $i (0 .. $min) {
                my $task = shift @$tasks;
                $before_work->($task) if $before_work;
                $ready[$i]->write($task);
            }
        } else {
            if (@ready == $num) {
                last;
            } else {
                if (my @written = $pipes->is_written) {
                    my @ready = $pipes->is_ready(@written);
                    for my $written (@ready) {
                        my $result = $written->read;
                        if ($after_work) {
                            $after_work->($result);
                        } else {
                            push @result, $result;
                        }
                    }
                } else {
                    die "unexpected";
                }
            }
        }
    }
    $pipes->close;
    $after_work ? 1 : @result;
}

1;
__END__