File Coverage

blib/lib/Parallel/Pipes/App.pm

Criterion	Covered	Total	%
statement	59	60	98.3
branch	18	28	64.2
condition			n/a
subroutine	8	8	100.0
pod	2	2	100.0
total	87	98	88.7

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

use Parallel::Pipes;

our $VERSION = '0.102';

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 $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;
                $after_work->($result) if $after_work;
            }
        }
        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;
                        $after_work->($result) if $after_work;
                    }
                } else {
                    die "unexpected";
                }
            }
        }
    }
    $pipes->close;
    1;
}

sub map :method {
    my ($class, %argv) = @_;

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

    my @task = map { [$_, $orig_tasks->[$_]] } 0..$#{$orig_tasks};
    my $work = sub {
        my ($index, $task) = @{$_[0]};
        my $result = $orig_work->($task);
        [$index, $result];
    };
    my @result;
    my $after_work = sub {
        my ($index, $result) = @{$_[0]};
        $result[$index] = $result;
    };
    $class->run(num => $orig_num, work => $work, tasks => \@task, after_work => $after_work);
    @result;
}

1;
__END__