File Coverage

blib/lib/Mojo/IOLoop/HoboProcess.pm

Criterion	Covered	Total	%
statement	30	80	37.5
branch	2	22	9.0
condition	0	13	0.0
subroutine	10	18	55.5
pod	3	4	75.0
total	45	137	32.8

###############################################################################
## ----------------------------------------------------------------------------
## Like Mojo::IOLoop::Subprocess, spawns subprocesses with MCE::Hobo instead.
##
###############################################################################

package Mojo::IOLoop::HoboProcess;

use strict;
use warnings;

use Mojo::Base -base;

our $VERSION = '0.005';

## no critic (BuiltinFunctions::ProhibitStringyEval)

use Carp 'croak';
use MCE::Hobo;
use Mojo::IOLoop;
use Mojo::IOLoop::Stream;
use Scalar::Util 'weaken';
use Socket qw( PF_UNIX PF_UNSPEC SOCK_STREAM );

BEGIN {
  $ENV{'MOJO_REACTOR'} ||= 'Mojo::Reactor::Poll' if ($^O eq 'MSWin32');

  unless (Mojo::IOLoop->can('hoboprocess')) {
    # Inject Mojo::IOLoop::hoboprocess, no impact to Mojo::IOLoop::subprocess

    eval '
      sub Mojo::IOLoop::hoboprocess {
        my $ioloop = shift;
        my $subprocess = Mojo::IOLoop::HoboProcess->new;

        weaken $subprocess->ioloop(
          ref $ioloop ? $ioloop : $ioloop->singleton
        )->{"ioloop"};

        return $subprocess->run(@_);
      }
    ';
  }
}

has ioloop => sub { Mojo::IOLoop->singleton };
has timeout => 0;

sub exit { my $self = shift->{'hobo'} || MCE::Hobo->self; $self->exit(@_) }

sub pid { shift->{'pid'} }

sub run {
  my ($self, $child, $parent) = @_;

  # Start the shared-manager process if not already started
  MCE::Shared->start;

  # Make socketpair or pipe for event notification
  my ($reader, $writer);

  if ($^O eq 'MSWin32') {
    socketpair($reader, $writer, PF_UNIX, SOCK_STREAM, PF_UNSPEC)
      or croak "Can't create socketpair: $!";
  }
  else {
    pipe($reader, $writer) or croak "Can't create pipe: $!";
  }

  $writer->autoflush(1);

  # Child
  my $hobo = MCE::Hobo->create({ posix_exit => 1 }, sub {
    close $reader;

    $self->ioloop->reset;
    $self->{'pid'} = MCE::Hobo->pid;

    my $results = eval { [ $self->$child ] } || [];
    my $error = $@; $error = '' if ($error eq "Hobo exited (0)\n");

    $error = "Hobo ". $self->{'pid'} ." exited abnormally"
      if ($error && $error =~ /^Hobo exited \(\S+\)$/);

    print $writer "done\n";
    close $writer;

    [ $error, @{ $results } ];
  });

  croak "Can't spawn Hobo: $!" unless defined($hobo);
  $self->{'hobo'} = $hobo, $self->{'pid'} = $hobo->pid;

  # Parent
  my $me     = $$;
  my $stream = Mojo::IOLoop::Stream->new($reader);

  $stream->timeout($self->timeout);
  $self->ioloop->stream($stream);

  $stream->on( read => sub {
    my $results = eval { $hobo->join };
    my $error = $hobo->error || shift @{ $results };
    $self->$parent($error, @{ $results });
  });

  $stream->on( timeout => sub {
    $hobo->kill('QUIT');
    eval { $hobo->join };
    $self->$parent( "Hobo ". $self->pid ." timed out", () );
  });

  $stream->on( close => sub {
    return unless $$ == $me;
    unless (exists $hobo->{'JOINED'}) {
      eval { $hobo->join };
      $self->$parent( "Hobo ". $self->{'pid'} ." exited abnormally", () );
    }
  });

  return $self;
}

1;

__END__