File Coverage

blib/lib/Parallel/Pipes.pm

Criterion	Covered	Total	%
statement	145	153	94.7
branch	34	46	73.9
condition	2	2	100.0
subroutine	29	29	100.0
pod	0	6	0.0
total	210	236	88.9

package Parallel::Pipes;
use 5.008001;
use strict;
use warnings;
use IO::Handle;
use IO::Select;

use constant WIN32 => $^O eq 'MSWin32';

our $VERSION = '0.201';

{
    package Parallel::Pipes::Impl;
    use Storable ();
    sub new {
        my ($class, %option) = @_;
        my $read_fh  = delete $option{read_fh}  or die;
        my $write_fh = delete $option{write_fh} or die;
        $write_fh->autoflush(1);
        bless { %option, read_fh => $read_fh, write_fh => $write_fh, buf => '' }, $class;
    }
    sub read :method {
        my $self = shift;
        my $_size = $self->_read(4) or return;
        my $size = unpack 'I', $_size;
        my $freezed = $self->_read($size);
        Storable::thaw($freezed);
    }
    sub write :method {
        my ($self, $data) = @_;
        my $freezed = Storable::freeze({data => $data});
        my $size = pack 'I', length($freezed);
        $self->_write("$size$freezed");
    }
    sub _read {
        my ($self, $size) = @_;
        my $fh = $self->{read_fh};
        my $offset = length $self->{buf};
        while ($offset < $size) {
            my $len = sysread $fh, $self->{buf}, 65536, $offset;
            if (!defined $len) {
                die $!;
            } elsif ($len == 0) {
                last;
            } else {
                $offset += $len;
            }
        }
        return substr $self->{buf}, 0, $size, '';
    }
    sub _write {
        my ($self, $data) = @_;
        my $fh = $self->{write_fh};
        my $size = length $data;
        my $offset = 0;
        while ($size) {
            my $len = syswrite $fh, $data, $size, $offset;
            if (!defined $len) {
                die $!;
            } elsif ($len == 0) {
                last;
            } else {
                $size   -= $len;
                $offset += $len;
            }
        }
        $size;
    }
}
{
    package Parallel::Pipes::Here;
    our @ISA = qw(Parallel::Pipes::Impl);
    use Carp ();
    sub new {
        my ($class, %option) = @_;
        $class->SUPER::new(%option, _written => 0);
    }
    sub is_written {
        my $self = shift;
        $self->{_written} == 1;
    }
    sub read :method {
        my $self = shift;
        if (!$self->is_written) {
            Carp::croak("This pipe has not been written; you cannot read it");
        }
        $self->{_written}--;
        return unless my $read = $self->SUPER::read;
        $read->{data};
    }
    sub write :method {
        my ($self, $task) = @_;
        if ($self->is_written) {
            Carp::croak("This pipe has already been written; you must read it first");
        }
        $self->{_written}++;
        $self->SUPER::write($task);
    }
}
{
    package Parallel::Pipes::There;
    our @ISA = qw(Parallel::Pipes::Impl);
}
{
    package Parallel::Pipes::Impl::NoFork;
    use Carp ();
    sub new {
        my ($class, %option) = @_;
        bless {%option}, $class;
    }
    sub is_written {
        my $self = shift;
        exists $self->{_result};
    }
    sub read :method {
        my $self = shift;
        if (!$self->is_written) {
            Carp::croak("This pipe has not been written; you cannot read it");
        }
        delete $self->{_result};
    }
    sub write :method {
        my ($self, $task) = @_;
        if ($self->is_written) {
            Carp::croak("This pipe has already been written; you must read it first");
        }
        my $result = $self->{work}->($task);
        $self->{_result} = $result;
    }
}

sub new {
    my ($class, $number, $work, $option) = @_;
    if (WIN32 and $number != 1) {
        die "The number of pipes must be 1 under WIN32 environment.\n";
    }
    my $self = bless {
        work => $work,
        number => $number,
        no_fork => $number == 1,
        pipes => {},
        option => $option || {},
    }, $class;

    if ($self->no_fork) {
        $self->{pipes}{-1} = Parallel::Pipes::Impl::NoFork->new(pid => -1, work => $self->{work});
    } else {
        $self->_fork for 1 .. $number;
    }
    $self;
}

sub no_fork { shift->{no_fork} }

sub _fork {
    my $self = shift;
    my $work = $self->{work};
    pipe my $read_fh1, my $write_fh1;
    pipe my $read_fh2, my $write_fh2;
    my $pid = fork;
    die "fork failed" unless defined $pid;
    if ($pid == 0) {
        srand;
        close $_ for $read_fh1, $write_fh2, map { ($_->{read_fh}, $_->{write_fh}) } $self->pipes;
        my $there = Parallel::Pipes::There->new(read_fh  => $read_fh2, write_fh => $write_fh1);
        while (my $read = $there->read) {
            $there->write( $work->($read->{data}) );
        }
        exit;
    }
    close $_ for $write_fh1, $read_fh2;
    $self->{pipes}{$pid} = Parallel::Pipes::Here->new(
        pid => $pid, read_fh => $read_fh1, write_fh => $write_fh2,
    );
}

sub pipes {
    my $self = shift;
    map { $self->{pipes}{$_} } sort { $a <=> $b } keys %{$self->{pipes}};
}

sub is_ready {
    my $self = shift;
    return $self->pipes if $self->no_fork;

    my @pipes = @_ ? @_ : $self->pipes;
    if (my @ready = grep { $_->{_written} == 0 } @pipes) {
        return @ready;
    }

    my $select = IO::Select->new(map { $_->{read_fh} } @pipes);
    my @ready;
    if (my $tick = $self->{option}{idle_tick}) {
        while (1) {
            if (my @r = $select->can_read($tick)) {
                @ready = @r;
                last;
            }
            $self->{option}{idle_work}->();
        }
    } else {
        @ready = $select->can_read;
    }

    my @return;
    for my $pipe (@pipes) {
        if (grep { $pipe->{read_fh} == $_ } @ready) {
            push @return, $pipe;
        }
    }
    return @return;
}

sub is_written {
    my $self = shift;
    grep { $_->is_written } $self->pipes;
}

sub close :method {
    my $self = shift;
    return if $self->no_fork;

    close $_ for map { ($_->{write_fh}, $_->{read_fh}) } $self->pipes;
    while (%{$self->{pipes}}) {
        my $pid = wait;
        if (delete $self->{pipes}{$pid}) {
            # OK
        } else {
            warn "wait() unexpectedly returns $pid\n";
        }
    }
}

1;
__END__