File Coverage

blib/lib/Proc/Forkmap.pm

Criterion	Covered	Total	%
statement	21	65	32.3
branch	0	14	0.0
condition	0	8	0.0
subroutine	7	13	53.8
pod	4	4	100.0
total	32	104	30.7

package Proc::Forkmap;
use POSIX qw(:sys_wait_h);
use Proc::Fork;
use Carp;
use IO::Pipe;
use strict;
use warnings;
use 5.010;

our $VERSION = '0.024';


sub new {
    my $class = shift;
    my $self = bless {@_}, $class;
    $self->_init;
    return $self;
}


sub _init {
    my $self = shift;
    ($self->{max_kids} //= 2) =~ /^[1-9]+$/
        or croak "max_kids value error";
    $self->{ipc} //= 0; #ipc off by default
}


sub max_kids {
    my ($self,$n) = @_;
    $n // return $self->{max_kids};
    ($n =~ /^[1-9]+$/)
        or croak "max_kids value error";
    $self->{max_kids} = $n;
}


sub ipc {
    my ($self,$n) = @_;
    $n // return $self->{ipc};
    $self->{ipc} = $n;
}


sub fmap {
    my ($self,$code) = (shift,shift);
    my %pids = ();
    my @rs = (); #result set of child return values
    my $max = $self->max_kids;
    my $ipc = $self->ipc;
    for my $proc (@_) {
        my $pipe = $ipc ? IO::Pipe->new : {}; #put this in your pipe, and smoke it
        #max kids?
        while ($max == keys %pids) {
            #free a spot in queue when a process completes
            for my $pid (keys %pids) {
                if (my $kid = waitpid($pid, WNOHANG)) {
                    delete $pids{$kid};
                    last;
                }
            }
        }

        run_fork { #processes fork here
            parent {
                $| = 1;
                my $kid = shift;
                $pids{$kid}++;
                if ($ipc) {
                    $pipe->reader();
                    while(<$pipe>) {
                        push @rs,$_;
                    }
                }
            }
            child {
                $| = 1;
                my $rs = $code->($proc);
                if ($ipc) {
                    $pipe->writer();
                    print $pipe $rs if defined $rs;
                }
                exit;
            }
            error {
                die "error: couldn't fork";
            }
        };
    }

    1 while (wait() != -1); #wait for the stragglers to finish
    return @rs;
}




1;

__END__