File Coverage

blib/lib/POE/Wheel/Run.pm

Criterion	Covered	Total	%
statement	377	562	67.0
branch	216	384	56.2
condition	77	157	49.0
subroutine	30	45	66.6
pod	20	21	95.2
total	720	1169	61.5

package POE::Wheel::Run;

use strict;

use vars qw($VERSION @ISA);
$VERSION = '1.365'; # NOTE - Should be #.### (three decimal places)

use Carp qw(carp croak);
use POSIX qw(
  sysconf setsid _SC_OPEN_MAX ECHO ICANON IEXTEN ISIG BRKINT ICRNL
  INPCK ISTRIP IXON CSIZE PARENB OPOST TCSANOW
);

use POE qw( Wheel Pipe::TwoWay Pipe::OneWay Driver::SysRW Filter::Line );
push @ISA, qw(POE::Wheel);

# http://rt.cpan.org/Ticket/Display.html?id=50068
# Avoid using these constants in Windows' subprocesses (actually
# interpreter threads).  Reported in the above ticket to avoid a
# memory leak.
my ($STD_INPUT_HANDLE, $STD_OUTPUT_HANDLE, $STD_ERROR_HANDLE);

BEGIN {
  die "$^O does not support fork()\n" if $^O eq 'MacOS';

  local $SIG{'__DIE__'} = 'DEFAULT';
  eval    { require IO::Pty; };
  if ($@) {
    eval '
      sub PTY_AVAILABLE () { 0 }
      sub TIOCSWINSZ_AVAILABLE () { 0 }
    ';
  }
  else {
    IO::Pty->import();
    eval 'sub PTY_AVAILABLE () { 1 }';

    eval { require IO::Tty; };
    if ($@) {
      eval 'sub TIOCSWINSZ_AVAILABLE () { 0 }';
    }
    else {
      IO::Tty->import('TIOCSWINSZ');
      eval 'sub TIOCSWINSZ_AVAILABLE () { 1 }';
    }
  }

  if (POE::Kernel::RUNNING_IN_HELL) {
    eval    { require Win32::Console; Win32::Console->import() };
    if ($@) { die "Win32::Console needed for POE::Wheel::Run on $^O:\n$@" }

    eval    {
      require Win32API::File;
      Win32API::File->import("FdGetOsFHandle");
    };
    if ($@) { die "Win32API::File needed for POE::Wheel::Run on $^O:\n$@" }

    eval    { require Win32::Process; Win32::Process->import() };
    if ($@) { die "Win32::Process needed for POE::Wheel::Run on $^O:\n$@" }

    eval    { require Win32::Job; Win32::Job->import() };
    if ($@) { die "Win32::Job needed for POE::Wheel::Run on $^O:\n$@" }

    eval    { require Win32; Win32->import() };
    if ($@) { die "Win32.pm needed for POE::Wheel::Run on $^O:\n$@" }

    $STD_INPUT_HANDLE  = STD_INPUT_HANDLE();
    $STD_OUTPUT_HANDLE = STD_OUTPUT_HANDLE();
    $STD_ERROR_HANDLE  = STD_ERROR_HANDLE();
  }

  # Determine the most file descriptors we can use.
  my $max_open_fds;
  eval {
    $max_open_fds = sysconf(_SC_OPEN_MAX);
  };
  $max_open_fds = 1024 unless $max_open_fds;
  eval "sub MAX_OPEN_FDS () { $max_open_fds }";
  die if $@;
};

# Offsets into $self.
sub UNIQUE_ID     () {  0 }
sub ERROR_EVENT   () {  1 }
sub CLOSE_EVENT   () {  2 }
sub PROGRAM       () {  3 }
sub CHILD_PID     () {  4 }
sub CONDUIT_TYPE  () {  5 }
sub IS_ACTIVE     () {  6 }
sub CLOSE_ON_CALL () {  7 }
sub STDIO_TYPE    () {  8 }

sub HANDLE_STDIN  () {  9 }
sub FILTER_STDIN  () { 10 }
sub DRIVER_STDIN  () { 11 }
sub EVENT_STDIN   () { 12 }
sub STATE_STDIN   () { 13 }
sub OCTETS_STDIN  () { 14 }

sub HANDLE_STDOUT () { 15 }
sub FILTER_STDOUT () { 16 }
sub DRIVER_STDOUT () { 17 }
sub EVENT_STDOUT  () { 18 }
sub STATE_STDOUT  () { 19 }

sub HANDLE_STDERR () { 20 }
sub FILTER_STDERR () { 21 }
sub DRIVER_STDERR () { 22 }
sub EVENT_STDERR  () { 23 }
sub STATE_STDERR  () { 24 }

sub MSWIN32_GROUP_PID () { 25 }

# Used to work around a bug in older perl versions.
sub CRIMSON_SCOPE_HACK ($) { 0 }

#------------------------------------------------------------------------------

sub new {
  my $type = shift;
  croak "$type needs an even number of parameters" if @_ & 1;
  my %params = @_;

  croak "wheels no longer require a kernel reference as their first parameter"
    if @_ and ref($_[0]) eq 'POE::Kernel';

  croak "$type requires a working Kernel" unless defined $poe_kernel;

  my $program = delete $params{Program};
  croak "$type needs a Program parameter" unless defined $program;

  my $prog_args = delete $params{ProgramArgs};
  $prog_args = [] unless defined $prog_args;
  croak "ProgramArgs must be an ARRAY reference"
    unless ref($prog_args) eq "ARRAY";

  my $priority_delta = delete $params{Priority};
  $priority_delta = 0 unless defined $priority_delta;

  my $close_on_call = delete $params{CloseOnCall};
  $close_on_call = 0 unless defined $close_on_call;

  my $user_id  = delete $params{User};
  my $group_id = delete $params{Group};

  # The following $stdio_type is new.  $conduit is kept around for now
  # to preserve the logic of the rest of the module.  This change
  # allows a Session using POE::Wheel::Run to define the type of pipe
  # to be created for stdin and stdout.  Read the POD on Conduit.
  # However, the documentation lies, because if Conduit is undefined,
  # $stdio_type is set to undefined (so the default pipe type provided
  # by POE::Pipe::TwoWay will be used). Otherwise, $stdio_type
  # determines what type of pipe Pipe:TwoWay creates unless it's
  # 'pty'.

  my $conduit = delete $params{Conduit};
  my $stdio_type;
  if (defined $conduit) {
    croak "$type\'s Conduit type ($conduit) is unknown" if (
      $conduit ne 'pipe' and
      $conduit ne 'pty'  and
      $conduit ne 'pty-pipe' and
      $conduit ne 'socketpair' and
      $conduit ne 'inet'
    );
    unless ($conduit =~ /^pty(-pipe)?$/) {
      $stdio_type = $conduit;
      $conduit = "pipe";
    }
  }
  else {
    $conduit = "pipe";
  }

  my $winsize = delete $params{Winsize};

  if ($winsize) {
    carp "winsize can only be specified for a Conduit of type pty"
      if $conduit !~ /^pty(-pipe)?$/ and $winsize;

    if( 'ARRAY' eq ref $winsize and 2==@$winsize ) {
        # Standard VGA cell in 9x16
        # http://en.wikipedia.org/wiki/VGA-compatible_text_mode#Fonts
        $winsize->[2] = $winsize->[1]*9;
        $winsize->[3] = $winsize->[0]*16;
    }
    carp "winsize must be a 4 element arrayref" unless ref($winsize) eq 'ARRAY'
      and scalar @$winsize == 4;

    carp "winsize only works when IO::Tty::TIOCSWINSZ is"
      unless TIOCSWINSZ_AVAILABLE;
  }

  my $stdin_event  = delete $params{StdinEvent};
  my $stdout_event = delete $params{StdoutEvent};
  my $stderr_event = delete $params{StderrEvent};

  if ($conduit eq 'pty' and defined $stderr_event) {
    carp "ignoring StderrEvent with pty conduit";
    undef $stderr_event;
  }

  #croak "$type needs at least one of StdinEvent, StdoutEvent or StderrEvent"
  #  unless (defined($stdin_event) or defined($stdout_event) or defined ($stderr_event));

  my $stdio_driver  = delete $params{StdioDriver}  || POE::Driver::SysRW->new();
  my $stdin_driver  = delete $params{StdinDriver}  || $stdio_driver;
  my $stdout_driver = delete $params{StdoutDriver} || $stdio_driver;
  my $stderr_driver = delete $params{StderrDriver} || POE::Driver::SysRW->new();

  my $stdio_filter  = delete $params{Filter};
  my $stdin_filter  = delete $params{StdinFilter};
  my $stdout_filter = delete $params{StdoutFilter};
  my $stderr_filter = delete $params{StderrFilter};

  #For optional redirection...
  my $redir_err     = delete $params{RedirectStderr};
  my $redir_out     = delete $params{RedirectStdout};
  my $redir_in      = delete $params{RedirectStdin};
  my $redir_output  = delete $params{RedirectOutput};

  my $no_stdin      = delete $params{NoStdin};

  if(defined $redir_output) {
    $redir_out = $redir_err = $redir_output;
  }

  #Sanity check. We can't wait for redirected filehandles
  if( (defined $redir_in and defined $stdin_event) ||
     (defined $redir_out and defined $stdout_event) ||
     (defined $redir_err and defined $stderr_event) ) {
    croak("Redirect* and *Event stdio options are mutually exclusive");
  }

  if (defined $stdio_filter) {
    croak "Filter and StdioFilter cannot be used together"
      if defined $params{StdioFilter};
    croak "Replace deprecated Filter with StdioFilter and StderrFilter"
      if defined $stderr_event and not defined $stderr_filter;
    carp "Filter is deprecated.  Please try StdioFilter and/or StderrFilter";
  }
  else {
    $stdio_filter = delete $params{StdioFilter};
  }
  $stdio_filter = POE::Filter::Line->new(Literal => "\n")
    unless defined $stdio_filter;

  $stdin_filter  = $stdio_filter unless defined $stdin_filter;
  $stdout_filter = $stdio_filter unless defined $stdout_filter;

  if ($conduit eq 'pty' and defined $stderr_filter) {
    carp "ignoring StderrFilter with pty conduit";
    undef $stderr_filter;
  }
  else {
    $stderr_filter = POE::Filter::Line->new(Literal => "\n")
      unless defined $stderr_filter;
  }

  croak "$type needs either StdioFilter or StdinFilter when using StdinEvent"
    if defined($stdin_event) and not defined($stdin_filter);
  croak "$type needs either StdioFilter or StdoutFilter when using StdoutEvent"
    if defined($stdout_event) and not defined($stdout_filter);
  croak "$type needs a StderrFilter when using StderrEvent"
    if defined($stderr_event) and not defined($stderr_filter);

  my $error_event = delete $params{ErrorEvent};
  my $close_event = delete $params{CloseEvent};

  my $no_setsid = delete $params{NoSetSid};
  my $no_setpgrp = delete $params{NoSetPgrp};

  # Make sure the user didn't pass in parameters we're not aware of.
  if (scalar keys %params) {
    carp(
      "unknown parameters in $type constructor call: ",
      join(', ', sort keys %params)
    );
  }

  # Did the user mangle stdio?
  unless (ref($program) eq 'CODE') {
    croak "Someone has closed or moved STDIN... exec() won't find it"
      unless defined fileno(STDIN) && fileno(STDIN) == 0;
    croak "Someone has closed or moved STDOUT... exec() won't find it"
      unless tied(*STDOUT) || defined fileno(STDOUT) && fileno(STDOUT) == 1;
    croak "Someone has closed or moved STDERR... exec() won't find it"
      unless tied(*STDERR) || defined fileno(STDERR) && fileno(STDERR) == 2;
  }

  my (
    $stdin_read, $stdout_write, $stdout_read, $stdin_write,
    $stderr_read, $stderr_write,
  );

  _filespec_to_fh(\$stdin_read, "<", $redir_in);
  if($redir_output) {
    _filespec_to_fh(\$stdout_write, ">", $redir_output);
    _filespec_to_fh(\$stderr_write, ">", $stdout_write);
  } else {
    _filespec_to_fh(\$stdout_write, ">", $redir_out);
    _filespec_to_fh(\$stderr_write, ">", $redir_err);
  }

  # Create a semaphore pipe.  This is used so that the parent doesn't
  # begin listening until the child's stdio has been set up.

  my ($sem_pipe_read, $sem_pipe_write) = POE::Pipe::OneWay->new();
  croak "could not create semaphore pipe: $!" unless defined $sem_pipe_read;

  # Use IO::Pty if requested.  IO::Pty turns on autoflush for us.

  if(defined $stdout_event
     or defined $stdin_event
     or defined $stderr_event
     or (!$no_stdin))
  #Bypass all the conduit handling if the user does not care for child I/O
  {
    if ($conduit =~ /^pty(-pipe)?$/) {
      croak "IO::Pty is not available" unless PTY_AVAILABLE;

      if(defined $redir_err or defined $redir_in or defined $redir_out) {
        croak "Redirection with pty conduit is unsupported";
      }

      $stdin_write = $stdout_read = IO::Pty->new();
      croak "could not create master pty: $!" unless defined $stdout_read;
      if ($conduit eq "pty-pipe") {
        ($stderr_read, $stderr_write) = POE::Pipe::OneWay->new();
        croak "could not make stderr pipes: $!"
          unless defined $stderr_read and defined $stderr_write;
      }
    }

    # Use pipes otherwise.
    elsif ($conduit eq 'pipe') {
      # We make more pipes than strictly necessary in case someone wants
      # to turn some on later.  Uses a TwoWay pipe for STDIN/STDOUT and
      # a OneWay pipe for STDERR.  This may save 2 filehandles if
      # socketpair() is available and no other $stdio_type is selected.

      foreach (
        [\$redir_out, \$stdout_read, \$stdout_write, $stdout_event, "stdout"],
        [\$redir_err, \$stderr_read, \$stderr_write, $stderr_event, "stderr"],
        [\$redir_in, \$stdin_read, \$stdin_write, $stdin_event, "stdin"]
        ) {
        my ($redir_ref,$rfd_ref,$wfd_ref,$evname, $prettyprint) = @$_;
        if(defined $evname && (!defined $$redir_ref)) {
          ($$rfd_ref,$$wfd_ref) = POE::Pipe::OneWay->new();
          croak "could not make $prettyprint pipe: $!"
            unless defined $$rfd_ref and defined $$wfd_ref;
        }
      }
      unless (defined($redir_in) or $no_stdin) {
        ($stdin_read, $stdin_write) = POE::Pipe::OneWay->new();
        croak "could not make stdin pipe $!"
          unless defined $stdin_write and defined $stdin_read;
      }
    }

    # Sanity check.
    else {
      croak "unknown conduit type $conduit";
    }
  }

  # Block signals until safe
  my $must_unmask;
  if( $poe_kernel->can( '_data_sig_mask_all' ) ) {
    $poe_kernel->_data_sig_mask_all;
    $must_unmask = 1;
  }

  # Fork!  Woo-hoo!
  my $pid = fork;

  # Child.  Parent side continues after this block.
  unless ($pid) {
    # removed the croak because it wasn't "safe" RT#56417
    #croak "couldn't fork: $!" unless defined $pid;
    # ANY OTHER DIE/CROAK/EXIT/WHATEVER in the child MUST use the helper!
    __PACKAGE__->_warn_and_exit_child( "couldn't fork: $!", int( $! ) )
      unless defined $pid;

    # Stdio should not be tied.  Resolves rt.cpan.org ticket 1648.
    if (tied *STDIN) {
      carp "Cannot redirect out of tied STDIN.  Untying it";
      untie *STDIN;
    }

    if (tied *STDOUT) {
      carp "Cannot redirect into tied STDOUT.  Untying it";
      untie *STDOUT;
    }

    if (tied *STDERR) {
      carp "Cannot redirect into tied STDERR.  Untying it";
      untie *STDERR;
    }

    # If running pty, we delay the slave side creation 'til after
    # doing the necessary bits to become our own [unix] session.
    if ($conduit =~ /^pty(-pipe)?$/) {

      # Become a new unix session.
      # Program 19.3, APITUE.  W. Richard Stevens built my hot rod.
      eval 'setsid()' unless $no_setsid;

      # Acquire a controlling terminal.  Program 19.3, APITUE.
      $stdin_write->make_slave_controlling_terminal();

      # Open the slave side of the pty.
      $stdin_read = $stdout_write = $stdin_write->slave();
      __PACKAGE__->_warn_and_exit_child( "could not create slave pty: $!", int( $! ) )
        unless defined $stdin_read;

      # For a simple pty conduit, stderr is wedged into stdout.
      $stderr_write = $stdout_write if $conduit eq 'pty';

      # Put the pty conduit (slave side) into "raw" or "cbreak" mode,
      # per APITUE 19.4 and 11.10.
      $stdin_read->set_raw();

      if (TIOCSWINSZ_AVAILABLE) {
        if ($winsize) {
          ioctl($stdin_read, TIOCSWINSZ, pack('vvvv', @$winsize));
        }
      }
      else {
        # Set the pty conduit (slave side) window size to our window
        # size.  APITUE 19.4 and 19.5.

        eval { $stdin_read->clone_winsize_from(\*STDIN) } if -T STDIN;
      }
    }
    else {
      # TODO - Can this be block eval?  Or a do{} block?
      eval 'setpgrp(0,0)' unless $no_setpgrp;
    }

    # Reset all signals in the child process.  POE's own handlers are
    # silly to keep around in the child process since POE won't be
    # using them.
    my @safe_signals = $poe_kernel->_data_sig_get_safe_signals();
    @SIG{@safe_signals} = ("DEFAULT") x @safe_signals;
    $poe_kernel->_data_sig_unmask_all if $must_unmask;

    # TODO How to pass events to the parent process?  Maybe over a
    # expedited (OOB) filehandle.

    # Fix the child process' priority.  Don't bother doing this if it
    # wasn't requested.  Can't emit events on failure because we're in
    # a separate process, so just fail quietly.

    if ($priority_delta) {
      eval {
        if (defined(my $priority = getpriority(0, $$))) {
          unless (setpriority(0, $$, $priority + $priority_delta)) {
            # TODO can't set child priority
          }
        }
        else {
          # TODO can't get child priority
        }
      };
      if ($@) {
        # TODO can't get child priority
      }
    }

    # Fix the group ID.  TODO Add getgrnam so group IDs can be
    # specified by name.  TODO Warn if not superuser to begin with.
    if (defined $group_id) {
      $( = $) = $group_id;
    }

    # Fix the user ID.  TODO Add getpwnam so user IDs can be specified
    # by name.  TODO Warn if not superuser to begin with.
    if (defined $user_id) {
      $< = $> = $user_id;
    }

    # Close what the child won't need.
    close $stdin_write if defined $stdin_write;
    close $stdout_read if defined $stdout_read;
    close $stderr_read if defined $stderr_read;

    if (POE::Kernel::RUNNING_IN_HELL) {
      __PACKAGE__->_redirect_child_stdio_in_hell(
        $stdin_read, $stdout_write, $stderr_write
      );
    }

    else {
      __PACKAGE__->_redirect_child_stdio_sanely(
        $stdin_read, $stdout_write, $stderr_write
      );
    }

    # Make STDOUT and/or STDERR auto-flush.
    select STDERR;  $| = 1;
    select STDOUT;  $| = 1;

    # The child doesn't need to read from the semaphore pipe.
    $sem_pipe_read = undef;

    # Run Perl code.  This is fairly consistent across most systems.

    if (ref($program) eq 'CODE') {

      # Tell the parent that the stdio has been set up.
      print $sem_pipe_write "go\n";
      close $sem_pipe_write;

      # Close any close-on-exec file descriptors.  Except STDIN,
      # STDOUT, and STDERR, of course.
      if ($close_on_call) {
        for (0..MAX_OPEN_FDS-1) {
          next if fileno(STDIN)  == $_;
          next if fileno(STDOUT) == $_;
          next if fileno(STDERR) == $_;
          POSIX::close($_);
        }
      }

      # TODO what if the program tries to exit? It needs to use
      # our _exit_child_any_way_we_can handler...
      # Should we replace CORE::exit? CORE::die too? blahhhhhh
      # We've documented that users should not do it, but who knows!
      eval { $program->(@$prog_args) };

      my $exitval;
      if ($@) {
        chomp $@;
        warn "$@\n";
        $exitval = -1;
      }

      __PACKAGE__->_exit_child_any_way_we_can( $exitval || 0 );
    }

    # Execute an external program.  This gets weird.

    # Windows! What I do for you!
    __PACKAGE__->_exec_in_hell(
      $close_on_call, $sem_pipe_write, $program, $prog_args
    ) if POE::Kernel::RUNNING_IN_HELL;

    # Everybody else seems sane.
    # Tell the parent that the stdio has been set up.
    print $sem_pipe_write "go\n";
    close $sem_pipe_write;

    # exec(ARRAY)
    if (ref($program) eq 'ARRAY') {
      exec(@$program, @$prog_args)
         or __PACKAGE__->_warn_and_exit_child(
           "can't exec (@$program) in child pid $$: $!", int( $! ) );
    }

    # exec(SCALAR)
    exec(join(" ", $program, @$prog_args))
      or __PACKAGE__->_warn_and_exit_child(
        "can't exec ($program) in child pid $$: $!", int( $! ) );
  }

  # Parent here.  Close what the parent won't need.

  defined($stdin_read)   and close $stdin_read;
  defined($stdout_write) and close $stdout_write;
  defined($stderr_write) and close $stderr_write;



  # Also close any slave ptys
  $stdout_read->close_slave() if (
    defined $stdout_read and ref($stdout_read) eq 'IO::Pty'
  );

  $stderr_read->close_slave() if (
    defined $stderr_read and ref($stderr_read) eq 'IO::Pty'
  );

  my $active_count = 0;
  $active_count++ if $stdout_event and $stdout_read;
  $active_count++ if $stderr_event and $stderr_read;

  my $self = bless [
    &POE::Wheel::allocate_wheel_id(),  # UNIQUE_ID
    $error_event,   # ERROR_EVENT
    $close_event,   # CLOSE_EVENT
    $program,       # PROGRAM
    $pid,           # CHILD_PID
    $conduit,       # CONDUIT_TYPE
    $active_count,  # IS_ACTIVE
    $close_on_call, # CLOSE_ON_CALL
    $stdio_type,    # STDIO_TYPE
    # STDIN
    $stdin_write,   # HANDLE_STDIN
    $stdin_filter,  # FILTER_STDIN
    $stdin_driver,  # DRIVER_STDIN
    $stdin_event,   # EVENT_STDIN
    undef,          # STATE_STDIN
    0,              # OCTETS_STDIN
    # STDOUT
    $stdout_read,   # HANDLE_STDOUT
    $stdout_filter, # FILTER_STDOUT
    $stdout_driver, # DRIVER_STDOUT
    $stdout_event,  # EVENT_STDOUT
    undef,          # STATE_STDOUT
    # STDERR
    $stderr_read,   # HANDLE_STDERR
    $stderr_filter, # FILTER_STDERR
    $stderr_driver, # DRIVER_STDERR
    $stderr_event,  # EVENT_STDERR
    undef,          # STATE_STDERR
    undef,          # MSWIN32_GROUP_PID
  ], $type;

  # PG- I suspect <> might need PIPE
  $poe_kernel->_data_sig_unmask_all if $must_unmask;

  # Wait here while the child sets itself up.
  $sem_pipe_write = undef;
  {
    local $/ = "\n";  # TODO - Needed?
    my $chldout = <$sem_pipe_read>;
    chomp $chldout;
    $self->[MSWIN32_GROUP_PID] = $chldout if $chldout ne 'go';
  }
  close $sem_pipe_read;

  $self->_define_stdin_flusher() if defined $stdin_write;
  $self->_define_stdout_reader() if defined $stdout_read;
  $self->_define_stderr_reader() if defined $stderr_read;

  return $self;
}

#------------------------------------------------------------------------------
# Define the internal state that will flush output to the child
# process' STDIN pipe.

sub _define_stdin_flusher {
  my $self = shift;

  # Read-only members.  If any of these change, then the write state
  # is invalidated and needs to be redefined.
  my $unique_id    = $self->[UNIQUE_ID];
  my $driver       = $self->[DRIVER_STDIN];
  my $error_event  = \$self->[ERROR_EVENT];
  my $close_event  = \$self->[CLOSE_EVENT];
  my $stdin_filter = $self->[FILTER_STDIN];
  my $stdin_event  = \$self->[EVENT_STDIN];
  my $is_active    = \$self->[IS_ACTIVE];

  # Read/write members.  These are done by reference, to avoid pushing
  # $self into the anonymous sub.  Extra copies of $self are bad and
  # can prevent wheels from destructing properly.
  my $stdin_octets = \$self->[OCTETS_STDIN];

  # Register the select-write handler.
  $poe_kernel->state(
    $self->[STATE_STDIN] = ref($self) . "($unique_id) -> select stdin",
    sub {                             # prevents SEGV
      0 && CRIMSON_SCOPE_HACK('<');
                                      # subroutine starts here
      my ($k, $me, $handle) = @_[KERNEL, SESSION, ARG0];

      $$stdin_octets = $driver->flush($handle);

      # When you can't write, nothing else matters.
      if ($!) {
        $$error_event && $k->call(
          $me, $$error_event,
          'write', ($!+0), $!, $unique_id, "STDIN"
        );
        $k->select_write($handle);
      }

      # Could write, or perhaps couldn't but only because the
      # filehandle's buffer is choked.
      else {

        # All chunks written; fire off a "flushed" event.
        unless ($$stdin_octets) {
          $k->select_pause_write($handle);
          $$stdin_event && $k->call($me, $$stdin_event, $unique_id);
        }
      }
    }
  );

  $poe_kernel->select_write($self->[HANDLE_STDIN], $self->[STATE_STDIN]);

  # Pause the write select immediately, unless output is pending.
  $poe_kernel->select_pause_write($self->[HANDLE_STDIN])
    unless ($self->[OCTETS_STDIN]);
}

#------------------------------------------------------------------------------
# Define the internal state that will read input from the child
# process' STDOUT pipe.  This is virtually identical to
# _define_stderr_reader, but they aren't implemented as a common
# function for speed reasons.

sub _define_stdout_reader {
  my $self = shift;

  # Can't do anything if we don't have a handle.
  return unless defined $self->[HANDLE_STDOUT];

  # No event?  Unregister the handler and leave.
  my $stdout_event  = \$self->[EVENT_STDOUT];
  unless ($$stdout_event) {
    $poe_kernel->select_read($self->[HANDLE_STDOUT]);
    return;
  }

  # If any of these change, then the read state is invalidated and
  # needs to be redefined.
  my $unique_id     = $self->[UNIQUE_ID];
  my $driver        = $self->[DRIVER_STDOUT];
  my $stdout_filter = $self->[FILTER_STDOUT];

  # These can change without redefining the callback since they're
  # enclosed by reference.
  my $is_active     = \$self->[IS_ACTIVE];
  my $close_event   = \$self->[CLOSE_EVENT];
  my $error_event   = \$self->[ERROR_EVENT];

  # Register the select-read handler for STDOUT.
  if (
    $stdout_filter->can("get_one") and
    $stdout_filter->can("get_one_start")
  ) {
    $poe_kernel->state(
      $self->[STATE_STDOUT] = ref($self) . "($unique_id) -> select stdout",
      sub {
        # prevents SEGV
        0 && CRIMSON_SCOPE_HACK('<');

        # subroutine starts here
        my ($k, $me, $handle) = @_[KERNEL, SESSION, ARG0];
        if (defined(my $raw_input = $driver->get($handle))) {
          $stdout_filter->get_one_start($raw_input);
          while (1) {
            my $next_rec = $stdout_filter->get_one();
            last unless @$next_rec;
            foreach my $cooked_input (@$next_rec) {
              $k->call($me, $$stdout_event, $cooked_input, $unique_id);
            }
          }
        }
        else {
          $$error_event and $k->call(
            $me, $$error_event,
            'read', ($!+0), $!, $unique_id, 'STDOUT'
          );
          unless (--$$is_active) {
            $k->call( $me, $$close_event, $unique_id )
              if defined $$close_event;
          }
          $k->select_read($handle);
        }
      }
    );
  }

  # Otherwise we can't get one.
  else {
    $poe_kernel->state(
      $self->[STATE_STDOUT] = ref($self) . "($unique_id) -> select stdout",
      sub {
        # prevents SEGV
        0 && CRIMSON_SCOPE_HACK('<');

        # subroutine starts here
        my ($k, $me, $handle) = @_[KERNEL, SESSION, ARG0];
        if (defined(my $raw_input = $driver->get($handle))) {
          foreach my $cooked_input (@{$stdout_filter->get($raw_input)}) {
            $k->call($me, $$stdout_event, $cooked_input, $unique_id);
          }
        }
        else {
          $$error_event and
            $k->call(
              $me, $$error_event,
              'read', ($!+0), $!, $unique_id, 'STDOUT'
            );
          unless (--$$is_active) {
            $k->call( $me, $$close_event, $unique_id )
              if defined $$close_event;
          }
          $k->select_read($handle);
        }
      }
    );
  }

  # register the state's select
  $poe_kernel->select_read($self->[HANDLE_STDOUT], $self->[STATE_STDOUT]);
}

#------------------------------------------------------------------------------
# Define the internal state that will read input from the child
# process' STDERR pipe.

sub _define_stderr_reader {
  my $self = shift;

  # Can't do anything if we don't have a handle.
  return unless defined $self->[HANDLE_STDERR];

  # No event?  Unregister the handler and leave.
  my $stderr_event  = \$self->[EVENT_STDERR];
  unless ($$stderr_event) {
    $poe_kernel->select_read($self->[HANDLE_STDERR]);
    return;
  }

  my $unique_id     = $self->[UNIQUE_ID];
  my $driver        = $self->[DRIVER_STDERR];
  my $stderr_filter = $self->[FILTER_STDERR];

  # These can change without redefining the callback since they're
  # enclosed by reference.
  my $error_event   = \$self->[ERROR_EVENT];
  my $close_event   = \$self->[CLOSE_EVENT];
  my $is_active     = \$self->[IS_ACTIVE];

  # Register the select-read handler for STDERR.
  if (
    $stderr_filter->can("get_one") and
    $stderr_filter->can("get_one_start")
  ) {
    $poe_kernel->state(
      $self->[STATE_STDERR] = ref($self) . "($unique_id) -> select stderr",
      sub {
        # prevents SEGV
        0 && CRIMSON_SCOPE_HACK('<');

        # subroutine starts here
        my ($k, $me, $handle) = @_[KERNEL, SESSION, ARG0];
        if (defined(my $raw_input = $driver->get($handle))) {
          $stderr_filter->get_one_start($raw_input);
          while (1) {
            my $next_rec = $stderr_filter->get_one();
            last unless @$next_rec;
            foreach my $cooked_input (@$next_rec) {
              $k->call($me, $$stderr_event, $cooked_input, $unique_id);
            }
          }
        }
        else {
          $$error_event and $k->call(
            $me, $$error_event,
            'read', ($!+0), $!, $unique_id, 'STDERR'
          );
          unless (--$$is_active) {
            $k->call( $me, $$close_event, $unique_id )
              if defined $$close_event;
          }
          $k->select_read($handle);
        }
      }
    );
  }

  # Otherwise we can't get_one().
  else {
    $poe_kernel->state(
      $self->[STATE_STDERR] = ref($self) . "($unique_id) -> select stderr",
      sub {
        # prevents SEGV
        0 && CRIMSON_SCOPE_HACK('<');

        # subroutine starts here
        my ($k, $me, $handle) = @_[KERNEL, SESSION, ARG0];
        if (defined(my $raw_input = $driver->get($handle))) {
          foreach my $cooked_input (@{$stderr_filter->get($raw_input)}) {
            $k->call($me, $$stderr_event, $cooked_input, $unique_id);
          }
        }
        else {
          $$error_event and $k->call(
            $me, $$error_event,
            'read', ($!+0), $!, $unique_id, 'STDERR'
          );
          unless (--$$is_active) {
            $k->call( $me, $$close_event, $unique_id )
              if defined $$close_event;
          }
          $k->select_read($handle);
        }
      }
    );
  }

  # Register the state's select.
  $poe_kernel->select_read($self->[HANDLE_STDERR], $self->[STATE_STDERR]);
}

#------------------------------------------------------------------------------
# Redefine events.

sub event {
  my $self = shift;
  push(@_, undef) if (scalar(@_) & 1);

  my ($redefine_stdin, $redefine_stdout, $redefine_stderr) = (0, 0, 0);

  while (@_) {
    my ($name, $event) = splice(@_, 0, 2);

    if ($name eq 'StdinEvent') {
      $self->[EVENT_STDIN] = $event;
      $redefine_stdin = 1;
    }
    elsif ($name eq 'StdoutEvent') {
      $self->[EVENT_STDOUT] = $event;
      $redefine_stdout = 1;
    }
    elsif ($name eq 'StderrEvent') {
      if ($self->[CONDUIT_TYPE] ne 'pty') {
        $self->[EVENT_STDERR] = $event;
        $redefine_stderr = 1;
      }
      else {
        carp "ignoring StderrEvent on a pty conduit";
      }
    }
    elsif ($name eq 'ErrorEvent') {
      $self->[ERROR_EVENT] = $event;
    }
    elsif ($name eq 'CloseEvent') {
      $self->[CLOSE_EVENT] = $event;
    }
    else {
      carp "ignoring unknown Run parameter '$name'";
    }
  }

  # Recalculate the active handles count.
  my $active_count = 0;
  $active_count++ if $self->[EVENT_STDOUT] and $self->[HANDLE_STDOUT];
  $active_count++ if $self->[EVENT_STDERR] and $self->[HANDLE_STDERR];
  $self->[IS_ACTIVE] = $active_count;
}

#------------------------------------------------------------------------------
# Destroy the wheel.

sub DESTROY {
  my $self = shift;

  return if(ref POE::Kernel->get_active_session eq 'POE::Kernel');

  # Turn off the STDIN thing.
  if ($self->[HANDLE_STDIN]) {
    $poe_kernel->select_write($self->[HANDLE_STDIN]);
    $self->[HANDLE_STDIN] = undef;
  }

  if ($self->[STATE_STDIN]) {
    $poe_kernel->state($self->[STATE_STDIN]);
    $self->[STATE_STDIN] = undef;
  }

  if ($self->[HANDLE_STDOUT]) {
    $poe_kernel->select_read($self->[HANDLE_STDOUT]);
    $self->[HANDLE_STDOUT] = undef;
  }
  if ($self->[STATE_STDOUT]) {
    $poe_kernel->state($self->[STATE_STDOUT]);
    $self->[STATE_STDOUT] = undef;
  }

  if ($self->[HANDLE_STDERR]) {
    $poe_kernel->select_read($self->[HANDLE_STDERR]);
    $self->[HANDLE_STDERR] = undef;
  }
  if ($self->[STATE_STDERR]) {
    $poe_kernel->state($self->[STATE_STDERR]);
    $self->[STATE_STDERR] = undef;
  }

  &POE::Wheel::free_wheel_id($self->[UNIQUE_ID]);
}

#------------------------------------------------------------------------------
# Queue input for the child process.

sub put {
  my ($self, @chunks) = @_;

  # Avoid big bada boom if someone put()s on a dead wheel.
  croak "Called put() on a wheel without an open STDIN handle" unless (
    $self->[HANDLE_STDIN]
  );

  if (
    $self->[OCTETS_STDIN] =  # assignment on purpose
    $self->[DRIVER_STDIN]->put($self->[FILTER_STDIN]->put(\@chunks))
  ) {
    $poe_kernel->select_resume_write($self->[HANDLE_STDIN]);
  }

  # No watermark.
  return 0;
}

#------------------------------------------------------------------------------
# Pause and resume various input events.

sub pause_stdout {
  my $self = shift;
  return unless defined $self->[HANDLE_STDOUT];
  $poe_kernel->select_pause_read($self->[HANDLE_STDOUT]);
}

sub pause_stderr {
  my $self = shift;
  return unless defined $self->[HANDLE_STDERR];
  $poe_kernel->select_pause_read($self->[HANDLE_STDERR]);
}

sub resume_stdout {
  my $self = shift;
  return unless defined $self->[HANDLE_STDOUT];
  $poe_kernel->select_resume_read($self->[HANDLE_STDOUT]);
}

sub resume_stderr {
  my $self = shift;
  return unless defined $self->[HANDLE_STDERR];
  $poe_kernel->select_resume_read($self->[HANDLE_STDERR]);
}

# Shutdown the pipe that leads to the child's STDIN.
sub shutdown_stdin {
  my $self = shift;
  return unless defined $self->[HANDLE_STDIN];

  $poe_kernel->select_write($self->[HANDLE_STDIN], undef);

  eval { local $^W = 0; shutdown($self->[HANDLE_STDIN], 1) };
  if ($@ or $self->[HANDLE_STDIN] != $self->[HANDLE_STDOUT]) {
    close $self->[HANDLE_STDIN];
  }

  $self->[HANDLE_STDIN] = undef;
}

#------------------------------------------------------------------------------
# Redefine filters, one at a time or at once.  This is based on PG's
# code in Wheel::ReadWrite.

sub _transfer_stdout_buffer {
  my ($self, $buf) = @_;

  my $old_output_filter = $self->[FILTER_STDOUT];

  # Assign old buffer contents to the new filter, and send out any
  # pending packets.

  # Use "get_one" if the new filter implements it.
  if (defined $buf) {
    if (
      $old_output_filter->can("get_one") and
      $old_output_filter->can("get_one_start")
    ) {
      $old_output_filter->get_one_start($buf);

      # Don't bother to continue if the filter has switched out from
      # under our feet again.  The new switcher will finish the job.

      while ($self->[FILTER_STDOUT] == $old_output_filter) {
        my $next_rec = $old_output_filter->get_one();
        last unless @$next_rec;
        foreach my $cooked_input (@$next_rec) {
          $poe_kernel->call(
            $poe_kernel->get_active_session(), $self->[EVENT_STDOUT],
            $cooked_input, $self->[UNIQUE_ID]
          );
        }
      }
    }

    # Otherwise use the old get() behavior.
    else {
      foreach my $cooked_input (@{$self->[FILTER_STDOUT]->get($buf)}) {
        $poe_kernel->call(
          $poe_kernel->get_active_session(), $self->[EVENT_STDOUT],
          $cooked_input, $self->[UNIQUE_ID]
        );
      }
    }
  }
}

sub _transfer_stderr_buffer {
  my ($self, $buf) = @_;

  my $old_output_filter = $self->[FILTER_STDERR];

  # Assign old buffer contents to the new filter, and send out any
  # pending packets.

  # Use "get_one" if the new filter implements it.
  if (defined $buf) {
    if (
      $old_output_filter->can("get_one") and
      $old_output_filter->can("get_one_start")
    ) {
      $old_output_filter->get_one_start($buf);

      # Don't bother to continue if the filter has switched out from
      # under our feet again.  The new switcher will finish the job.

      while ($self->[FILTER_STDERR] == $old_output_filter) {
        my $next_rec = $old_output_filter->get_one();
        last unless @$next_rec;
        foreach my $cooked_input (@$next_rec) {
          $poe_kernel->call(
            $poe_kernel->get_active_session(), $self->[EVENT_STDERR],
            $cooked_input, $self->[UNIQUE_ID]
          );
        }
      }
    }

    # Otherwise use the old get() behavior.
    else {
      foreach my $cooked_input (@{$self->[FILTER_STDERR]->get($buf)}) {
        $poe_kernel->call(
          $poe_kernel->get_active_session(), $self->[EVENT_STDERR],
          $cooked_input, $self->[UNIQUE_ID]
        );
      }
    }
  }
}

sub set_stdio_filter {
  my ($self, $new_filter) = @_;
  $self->set_stdout_filter($new_filter);
  $self->set_stdin_filter($new_filter);
}

sub set_stdin_filter {
  my ($self, $new_filter) = @_;
  $self->[FILTER_STDIN] = $new_filter;
}

sub set_stdout_filter {
  my ($self, $new_filter) = @_;

  my $buf = $self->[FILTER_STDOUT]->get_pending();
  $self->[FILTER_STDOUT] = $new_filter;

  $self->_transfer_stdout_buffer($buf);
}

sub set_stderr_filter {
  my ($self, $new_filter) = @_;

  my $buf = $self->[FILTER_STDERR]->get_pending();
  $self->[FILTER_STDERR] = $new_filter;

  $self->_transfer_stderr_buffer($buf);
}

sub get_stdin_filter {
  my $self = shift;
  return $self->[FILTER_STDIN];
}

sub get_stdout_filter {
  my $self = shift;
  return $self->[FILTER_STDOUT];
}

sub get_stderr_filter {
  my $self = shift;
  return $self->[FILTER_STDERR];
}

#------------------------------------------------------------------------------
# Data accessors.

sub get_driver_out_octets {
  $_[0]->[OCTETS_STDIN];
}

sub get_driver_out_messages {
  $_[0]->[DRIVER_STDIN]->get_out_messages_buffered();
}

sub ID {
  $_[0]->[UNIQUE_ID];
}

sub PID {
  $_[0]->[CHILD_PID];
}

sub kill {
  my ($self, $signal) = @_;
  $signal = 'TERM' unless defined $signal;
  if ( $self->[MSWIN32_GROUP_PID] ) {
    # TODO use https://rt.cpan.org/Ticket/Display.html?id=67774 when available :)
    Win32::Process::KillProcess( $self->[MSWIN32_GROUP_PID], 293 ) ? 1 : 0;
  }
  else {
    eval { kill $signal, $self->[CHILD_PID] };
  }
}

### Internal helpers.

sub _redirect_child_stdio_in_hell {
  my ($class, $stdin_read, $stdout_write, $stderr_write) = @_;

  # Win32 needs the stdio handles closed before they're reopened
  # because the standard handles aren't dup()'d.

  close STDIN;
  close STDOUT;
  close STDERR;

  $class->_redirect_child_stdio_sanely(
    $stdin_read, $stdout_write, $stderr_write
  );

  # The Win32 pseudo fork sets up the std handles in the child
  # based on the true win32 handles.  The reopening of stdio
  # handles isn't enough.  We must also set the underlying
  # Win32 notion of these handles for completeness.
  #
  # Only necessary for the exec, as Perl CODE subroutine goes
  # through 0/1/2 which are correct.  But of course that coderef
  # might invoke exec, so better do it regardless.
  #
  # HACK: Using Win32::Console as nothing else exposes
  # SetStdHandle
  #
  # TODO - https://rt.cpan.org/Ticket/Display.html?id=50068 claims
  # that these _SetStdHandle() calls may leak memory.  Do we have
  # alternatives?

  Win32::Console::_SetStdHandle(
    $STD_INPUT_HANDLE,
    FdGetOsFHandle(fileno($stdin_read))
  ) if defined $stdin_read;

  Win32::Console::_SetStdHandle(
    $STD_OUTPUT_HANDLE,
    FdGetOsFHandle(fileno($stdout_write))
  ) if defined $stdout_write;

  Win32::Console::_SetStdHandle(
    $STD_ERROR_HANDLE,
    FdGetOsFHandle(fileno($stderr_write))
  ) if defined $stderr_write;
}

sub _filespec_to_fh {
  my ($dest,$mode,$fspec) = @_;
  return unless defined $fspec;
  if(ref $fspec) {
    if (ref $fspec eq 'GLOB') {
      open $$dest, "$mode&", $fspec;
    } else {
      die("Bad file specifier '$fspec'");
    }
  } else {
    open $$dest, $mode, $fspec;
  }
}

sub _redirect_child_stdio_sanely {
  my ($class, $stdin_read, $stdout_write, $stderr_write) = @_;

  # Note: we use 2-arg open() below because Perl 5.6 doesn't recognize
  # the '>&' and '<&' modes with a 3-arg open()

  # Redirect STDIN from the read end of the stdin pipe.
  if(defined $stdin_read) {
    open( STDIN, "<&" . fileno($stdin_read) )
      or $class->_warn_and_exit_child(
        "can't redirect STDIN in child pid $$: $!", int( $! ) );
  }

  # Redirect STDOUT to the write end of the stdout pipe.
  if(defined $stdout_write) {
    open( STDOUT, ">&" . fileno($stdout_write) )
      or $class->_warn_and_exit_child(
        "can't redirect stdout in child pid $$: $!", int( $! ) );
  }
    # Redirect STDERR to the write end of the stderr pipe.
  if(defined $stderr_write) {
    open( STDERR, ">&" . fileno($stderr_write) )
      or $class->_warn_and_exit_child(
        "can't redirect stderr in child pid $$: $!", int( $! ) );
  }
}

sub _exit_child_any_way_we_can {
  my $class = shift;
  my $exitval = shift || 0;

  # First make sure stdio are flushed.
  close STDIN  if defined fileno(STDIN); # Voodoo?
  close STDOUT if defined fileno(STDOUT);
  close STDERR if defined fileno(STDERR);

  # On Windows, subprocesses run in separate threads.  All the "fancy"
  # methods act on entire processes, so they also exit the parent.

  unless (POE::Kernel::RUNNING_IN_HELL) {
    # Try to avoid triggering END blocks and object destructors.
    eval { POSIX::_exit( $exitval ); };

    # TODO those methods will not exit with $exitval... what to do?
    eval { CORE::kill KILL => $$; };
    eval { exec("$^X -e 0"); };
  } else {
    eval { CORE::kill( KILL => $$ ); };

    # TODO Interestingly enough, the KILL is not enough to terminate this process...
    # However, it *is* enough to stop execution of END blocks/etc
    # So we will end up falling through to the exit( $exitval ) below
  }

  # Do what we must.
  exit( $exitval );
}

# RUNNING_IN_HELL use Win32::Process to create a pucker new shiny
# process. It'll inherit our processes handles which is neat.

sub _exec_in_hell {
  my (
    $class, $close_on_call, $sem_pipe_write,
    $program, $prog_args
  ) = @_;

  # Close any close-on-exec file descriptors.
  # Except STDIN, STDOUT, and STDERR, of course.

  if ($close_on_call) {
    for (0..MAX_OPEN_FDS-1) {
      next if fileno(STDIN) == $_;
      next if fileno(STDOUT) == $_;
      next if fileno(STDERR) == $_;
      POSIX::close($_);
    }
  }

  my ($appname, $cmdline);

  if (ref $program eq 'ARRAY') {
    $appname = $program->[0];
    $cmdline = join(
      ' ',
      map { /\s/ && ! /"/ ? qq{"$_"} : $_ }
      (@$program, @$prog_args)
    );
  }
  else {
    $appname = undef;
    $cmdline = join(
      ' ', $program,
      map { /\s/ && ! /"/ ? qq{"$_"} : $_ }
      @$prog_args
    );
  }

  my $w32job;

  unless ( $w32job = Win32::Job->new() ) {
    print $sem_pipe_write "go\n\n"; # TODO why the double newline?
    close $sem_pipe_write;
    $class->_warn_and_exit_child(
      Win32::FormatMessage( Win32::GetLastError() ), Win32::GetLastError() );
  }

  my $w32pid;

  unless ( $w32pid = $w32job->spawn( $appname, $cmdline ) ) {
    print $sem_pipe_write "go\n";
    close $sem_pipe_write;
    $class->_warn_and_exit_child(
      Win32::FormatMessage( Win32::GetLastError() ), Win32::GetLastError() );
  }

  print $sem_pipe_write "$w32pid\n";
  close $sem_pipe_write;

  # TODO why 60? Why not MAX_INT so we don't do unnecessary work?
  my $ok = $w32job->watch( sub { 0 }, 60 );
  my $hashref = $w32job->status();

  # In case flushing them wasn't good enough.
  close STDOUT if defined fileno(STDOUT);
  close STDERR if defined fileno(STDERR);

  $class->_exit_child_any_way_we_can( $hashref->{$w32pid}->{exitcode} );
}

# Simple helper to ease the pain of warn+exit
sub _warn_and_exit_child {
  my( $class, $warning, $exitval ) = @_;

  warn "$warning\n";

  $class->_exit_child_any_way_we_can( $exitval );
}

1;

__END__