File Coverage

blib/lib/Parallel/Pipes/App.pm

Criterion	Covered	Total	%
statement	77	78	98.7
branch	22	32	68.7
condition			n/a
subroutine	9	9	100.0
pod	2	2	100.0
total	110	121	90.9

package Parallel::Pipes::App v1.0.0;
use v5.24;
use warnings;
use experimental qw(lexical_subs signatures);

use Parallel::Pipes;

sub _min ($x, $y) { $x < $y ? $x : $y }

sub run ($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 $init_work = $argv{init_work};
    my $idle_tick = $argv{idle_tick};
    my $idle_work = $argv{idle_work};

    my $pipes = Parallel::Pipes->new(
        $num,
        $work,
        $idle_tick ? { idle_tick => $idle_tick, idle_work => $idle_work } : (),
    );
    $init_work->($pipes) if $init_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, $written) if $after_work;
            }
        }
        if ($tasks->@*) {
            my $min = _min $tasks->$#*, $#ready;
            for my $i (0 .. $min) {
                my $task = shift $tasks->@*;
                $before_work->($task, $ready[$i]) 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, $written) if $after_work;
                    }
                } else {
                    die "unexpected";
                }
            }
        }
    }
    $pipes->close;
    1;
}

sub map ($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 ($item) {
        my ($index, $task) = $item->@*;
        my $result = $orig_work->($task);
        [$index, $result];
    };
    my @result;
    my $after_work = sub ($item, @) {
        my ($index, $result) = $item->@*;
        $result[$index] = $result;
    };
    $class->run(num => $orig_num, work => $work, tasks => \@task, after_work => $after_work);
    @result;
}

1;
__END__