File Coverage

blib/lib/Parallel/SubArray.pm

Criterion	Covered	Total	%
statement	55	55	100.0
branch	12	14	85.7
condition	4	5	80.0
subroutine	5	5	100.0
pod	1	1	100.0
total	77	80	96.2

package Parallel::SubArray;
require v5.8.6;
our $VERSION = 0.6;
use strict;
use Storable qw(store_fd fd_retrieve);
use Exporter 'import';
our @EXPORT_OK = qw(par);

sub par {
  my( $timeout ) = @_;
  sub {
    my @subs = @{ shift(@_) };
    my %rets;
    my $c = 0;
    for my $sub ( @subs ) {
      $c++;
      my( $parent_w, $child_r );
      pipe( $child_r, $parent_w );
      select((select($parent_w ), $| = 1)[0]);
      if( my $pid = fork ) {
	close $parent_w;
	$rets{ $pid } = { fd  => $child_r,
			  ord => $c
			};
      } else {
	die "Cannot fork: $!" unless defined $pid;
	close $child_r;
	my $exit = sub {
	  my( $save_val, $exit_val ) = @_;
	  store_fd $save_val, $parent_w;
	  close $parent_w;
	  exit $exit_val;
	};
	local $SIG{ALRM} = sub { $exit->( ['TIMEOUT'], 1 ) };
	alarm( $timeout || 0 );
	my $ret = eval{ $sub->() };
	my $err = $@;
	alarm( 0 );
	$exit->( [$err], 1 ) if $err;
	$exit->(  $ret , 0 );
      }
    }
    while(1) {
      my $pid = wait();
      my $err = $?;
      last if( $pid == -1 or not @subs );
      next if not exists $rets{ $pid };
      if( $err ) {
	$rets{ $pid }->{err} = fd_retrieve( $rets{ $pid }->{fd} )->[0];
      } else {
	$rets{ $pid }->{val} = fd_retrieve( $rets{ $pid }->{fd} );
      }
      close $rets{ $pid }->{fd};
      pop @subs;
    }
    my $r = sub {
      my( $key ) = @_;
      # can be optimized
      [ map  { $rets{$_}->{ $key } }
	sort { $rets{$a}->{ord} <=> $rets{$b}->{ord} }
	keys %rets
      ];
    };
    return wantarray ? ( $r->('val'), $r->('err') ) : $r->('val');
  }
}

1;
__END__