lib/Object/Remote/Role/Connector/PerlInterpreter.pm

   1 package Object::Remote::Role::Connector::PerlInterpreter;
   2
   3 use IPC::Open2;
   4 use IPC::Open3;
   5 use IO::Handle;
   6 use Symbol;
   7 use Object::Remote::Logging qw( :log :dlog );
   8 use Object::Remote::ModuleSender;
   9 use Object::Remote::Handle;
  10 use Object::Remote::Future;
  11 use Scalar::Util qw(blessed weaken);
  12 use Moo::Role;
  13
  14 with 'Object::Remote::Role::Connector';
  15
  16 has module_sender => (is => 'lazy');
  17
  18 #if no child_stderr file handle is specified then stderr
  19 #of the child will be connected to stderr of the parent
  20 has stderr => ( is => 'rw', default => sub { undef } );
  21
  22 sub _build_module_sender {
  23   my ($hook) =
  24     grep {blessed($_) && $_->isa('Object::Remote::ModuleLoader::Hook') }
  25       @INC;
  26   return $hook ? $hook->sender : Object::Remote::ModuleSender->new;
  27 }
  28
  29 has perl_command => (is => 'lazy');
  30 has watchdog_timeout => ( is => 'ro', required => 1, default => sub { 0 } );
  31
  32 #TODO convert the ulimit and nice values into configurable attributes
  33 sub _build_perl_command {[ 'sh -c "ulimit -v 200000; nice -n 15 perl -"' ] }
  34
  35 around connect => sub {
  36   my ($orig, $self) = (shift, shift);
  37   my $f = $self->$start::start($orig => @_);
  38   return future {
  39     $f->on_done(sub {
  40       my ($conn) = $f->get;
  41       $self->_setup_watchdog_reset($conn);
  42       my $sub = $conn->remote_sub('Object::Remote::Logging::init_logging_forwarding');
  43       $sub->('Object::Remote::Logging', Object::Remote::Logging->arg_router);
  44       Object::Remote::Handle->new(
  45         connection => $conn,
  46         class => 'Object::Remote::ModuleLoader',
  47         args => { module_sender => $self->module_sender }
  48       )->disarm_free;
  49       require Object::Remote::Prompt;
  50       Object::Remote::Prompt::maybe_set_prompt_command_on($conn);
  51     });
  52     $f;
  53   } 2;
  54 };
  55
  56 sub final_perl_command { shift->perl_command }
  57
  58 sub _start_perl {
  59   my $self = shift;
  60   my $given_stderr = $self->stderr;
  61   my $foreign_stderr;
  62
  63   Dlog_debug { "invoking connection to perl interpreter using command line: $_" } @{$self->final_perl_command};
  64
  65   if (defined($given_stderr)) {
  66       #if the stderr data goes to an existing file handle
  67       #an need an anonymous file handle is required
  68       #as the other half of a pipe style file handle pair
  69       #so the file handles can go into the run loop
  70       $foreign_stderr = gensym();
  71   } else {
  72       #if no file handle has been specified
  73       #for the child's stderr then connect
  74       #the child stderr to the parent stderr
  75       $foreign_stderr = ">&STDERR";
  76   }
  77
  78   my $pid = open3(
  79     my $foreign_stdin,
  80     my $foreign_stdout,
  81     $foreign_stderr,
  82     @{$self->final_perl_command},
  83   ) or die "Failed to run perl at '$_[0]': $!";
  84
  85   if (defined($given_stderr)) {
  86       Dlog_debug { "Child process STDERR is being handled via run loop" };
  87
  88       Object::Remote->current_loop
  89                     ->watch_io(
  90                         handle => $foreign_stderr,
  91                         on_read_ready => sub {
  92                           my $buf = '';
  93                           my $len = sysread($foreign_stderr, $buf, 32768);
  94                           if (!defined($len) or $len == 0) {
  95                             log_trace { "Got EOF or error on child stderr, removing from watcher" };
  96                             $self->stderr(undef);
  97                             Object::Remote->current_loop
  98                                           ->unwatch_io(
  99                                               handle => $foreign_stderr,
 100                                               on_read_ready => 1
 101                                             );
 102                           } else {
 103                               Dlog_trace { "got $len characters of stderr data for connection" };
 104                               print $given_stderr $buf or die "could not send stderr data: $!";
 105                           }
 106                          }
 107                       );
 108   }
 109
 110   return ($foreign_stdin, $foreign_stdout, $pid);
 111 }
 112
 113 sub _open2_for {
 114   my $self = shift;
 115   my ($foreign_stdin, $foreign_stdout, $pid) = $self->_start_perl(@_);
 116   my $to_send = $self->fatnode_text;
 117   log_debug { my $len = length($to_send); "Sending contents of fat node to remote node; size is '$len' characters"  };
 118   Object::Remote->current_loop
 119                 ->watch_io(
 120                     handle => $foreign_stdin,
 121                     on_write_ready => sub {
 122                       my $len = syswrite($foreign_stdin, $to_send, 32768);
 123                       if (defined $len) {
 124                         substr($to_send, 0, $len) = '';
 125                       }
 126                       # if the stdin went away, we'll never get Shere
 127                       # so it's not a big deal to simply give up on !defined
 128                       if (!defined($len) or 0 == length($to_send)) {
 129                         log_trace { "Got EOF or error when writing fatnode data to filehandle, unwatching it" };
 130                         Object::Remote->current_loop
 131                                       ->unwatch_io(
 132                                           handle => $foreign_stdin,
 133                                           on_write_ready => 1
 134                                         );
 135                       } else {
 136                           log_trace { "Sent $len bytes of fatnode data to remote side" };
 137                       }
 138                     }
 139                   );
 140   return ($foreign_stdin, $foreign_stdout, $pid);
 141 }
 142
 143 sub _setup_watchdog_reset {
 144     my ($self, $conn) = @_;
 145     my $timer_id;
 146
 147     return unless $self->watchdog_timeout;
 148
 149     Dlog_trace { "Creating Watchdog management timer for connection id $_" } $conn->_id;
 150
 151     weaken($conn);
 152
 153     $timer_id = Object::Remote->current_loop->watch_time(
 154         every => $self->watchdog_timeout / 3,
 155         code => sub {
 156             unless(defined($conn)) {
 157                 log_trace { "Weak reference to connection in Watchdog was lost, terminating update timer $timer_id" };
 158                 Object::Remote->current_loop->unwatch_time($timer_id);
 159                 return;
 160             }
 161
 162             Dlog_trace { "Reseting Watchdog for connection id $_" } $conn->_id;
 163             #we do not want to block in the run loop so send the
 164             #update off and ignore any result, we don't need it
 165             #anyway
 166             $conn->send_class_call(0, 'Object::Remote::WatchDog', 'reset');
 167         }
 168     );
 169 }
 170
 171 sub fatnode_text {
 172   my ($self) = @_;
 173   my $text = '';
 174
 175   require Object::Remote::FatNode;
 176
 177   if (defined($self->watchdog_timeout)) {
 178     $text = "my \$WATCHDOG_TIMEOUT = '" . $self->watchdog_timeout . "';\n";
 179     $text .= "alarm(\$WATCHDOG_TIMEOUT);\n";
 180   } else {
 181       $text = "my \$WATCHDOG_TIMEOUT = undef;\n";
 182   }
 183
 184   $text .= 'BEGIN { $ENV{OBJECT_REMOTE_DEBUG} = 1 }'."\n"
 185     if $ENV{OBJECT_REMOTE_DEBUG};
 186   $text .= <<'END';
 187 $INC{'Object/Remote/FatNode.pm'} = __FILE__;
 188 $Object::Remote::FatNode::DATA = <<'ENDFAT';
 189 END
 190   $text .= do { no warnings 'once'; $Object::Remote::FatNode::DATA };
 191   $text .= "ENDFAT\n";
 192   $text .= <<'END';
 193 eval $Object::Remote::FatNode::DATA;
 194 die $@ if $@;
 195 END
 196
 197   $text .= "__END__\n";
 198   return $text;
 199 }
 200
 201 1;