File Coverage

blib/lib/Proc/Topus.pm

Criterion	Covered	Total	%
statement	108	108	100.0
branch	48	56	85.7
condition	9	9	100.0
subroutine	13	13	100.0
pod	1	1	100.0
total	179	187	95.7

package Proc::Topus;

use strict;
use warnings;

use Exporter;
use Socket;


our $VERSION = '0.02';

our @ISA       = qw( Exporter );
our @EXPORT_OK = qw( spawn );


sub _CONDUIT () { 'socketpair' }
sub _PARENT  () { 0 }
sub _WORKER  () { 1 }
sub _READER  () { 0 }
sub _WRITER  () { 1 }


sub spawn {
  my $args = _args( @_ );

  my $workers = $args->{workers};
  die 'No workers defined'
    unless defined $workers && keys %$workers > 0;

  my $pairs = _alloc_pairs( $workers, $args->{conduit}, $args->{autoflush} );

  my %pids;
  for my $name ( keys %$workers ) {
    my $config = $workers->{$name};

    my $pid = fork;
    defined $pid
      or die 'fork: ', $!;

    $pids{$pid} = $name;
    next
      if $pid != 0;

    my @pairs = map { $_->[_WORKER] } @{ delete $pairs->{$name} };
    delete $pairs->{$_}
      for keys %$pairs;

    if( $config->{setsid} ) {
      require POSIX;
      POSIX::setsid()
        or die 'setsid: ', $!;
    }

    my $loader = $config->{loader};
    $loader->( @pairs )
      if ref $loader eq 'CODE';

    while( @pairs ) {
      my $pair = shift @pairs;

      my $pid = fork;
      defined $pid
        or die 'fork: ', $!;

      next
        if $pid != 0;

      splice @pairs;

      my $main = $config->{main};
      $main = sub { $pair }
        unless ref $main eq 'CODE';

      return $main->( @$pair[_READER, _WRITER] );
    }

    exit 0;
  }

  for my $name ( keys %$pairs ) {
    $pairs->{$name} = [
      map { { reader => $_->[_READER], writer => $_->[_WRITER] } }
        map { $_->[_PARENT] }
          @{ $pairs->{$name} }
    ];
  }

  _wait( \%pids );

  my $main = $args->{main};
  $main = sub { $pairs }
    unless ref $main eq 'CODE';

  $main->( $pairs )
}

sub _args {
  if( scalar @_ == 1 ) {
    unless( defined $_[0] && ref $_[0] eq 'HASH' ) {
      die "Single arguments must be a HASH ref";
    }

    return $_[0]
  }
  elsif( @_ % 2 ) {
    die "Odd number of arguments";
  }
  else {
    return { @_ };
  }
}

sub _alloc_pairs {
  my ( $workers, $_conduit, $_autoflush ) = @_;

  $_conduit = _CONDUIT
    unless defined $_conduit;

  $_autoflush = 1
    unless defined $_autoflush;

  my %pairs;
  for my $name ( keys %$workers ) {
    my $config = $workers->{$name};

    my $count = $config->{count};
    die "Invalid worker count ($name)"
      unless defined $count && $count > 0;

    my $conduit = $config->{conduit};
    $conduit = $_conduit
      unless defined $conduit;

    my $autoflush = $config->{autoflush};
    $autoflush = $_autoflush
      unless defined $autoflush;

    $pairs{$name} =
      [ map { _alloc_pair( $name, $conduit, $autoflush ) } 1 .. $count ];
  }

  \%pairs
}

sub _alloc_pair {
  my ( $name, $conduit, $autoflush ) = @_;

  if( $conduit eq 'socketpair' ) {
    socketpair my $ps, my $ws, AF_UNIX, SOCK_STREAM, PF_UNSPEC
      or die 'socketpair: ', $!;

    _autoflush( $autoflush, $ps, $ws );

    return _pair( $ps, $ps, $ws, $ws );
  }
  elsif( $conduit eq 'pipe' ) {
    pipe my $pr, my $ww
      or die 'pipe: ', $!;
    pipe my $wr, my $pw
      or die 'pipe: ', $!;

    _autoflush( $autoflush, $pr, $pw, $wr, $ww );

    return _pair( $pr, $pw, $wr, $ww );
  }

  die "Invalid worker conduit '$conduit' for worker '$name'";
}

sub _autoflush {
  return
    unless shift;

  my $fh = select;
  do { select $_; $| = 1 }
    for @_;
  select $fh
}

sub _pair {
  my ( $pr, $pw, $wr, $ww ) = @_;

  my $pp; @$pp[_READER, _WRITER] = ( $pr, $pw );
  my $wp; @$wp[_READER, _WRITER] = ( $wr, $ww );

  my $pair; @$pair[_PARENT, _WORKER] = ( $pp, $wp );
  $pair
}

sub _wait {
  my ( $pids ) = @_;

  my $errors = 0;

  until( keys %$pids == 0 ) {
    my $pid = waitpid -1, 0;
    die 'waitpid: ', $!
      if $pid == -1;

    my $name = delete $pids->{$pid};
    next
      unless defined $name;

    my $rc = $? >> 8;
    $errors++
      unless $rc == 0;
  }

  die "One or more loaders failed"
    if $errors;
}


1
__END__