X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?p=scpubgit%2FObject-Remote.git;a=blobdiff_plain;f=lib%2FObject%2FRemote%2FRole%2FConnector%2FPerlInterpreter.pm;h=31e13e2259cd96e4c3339e719c83a1eb015f4b64;hp=d418c236508b977b6b616c7de0465f60c03ea64a;hb=b9baacc29444767f88abdbca93f65c8bd5e5a676;hpb=4a9fa1a5a329754549f998536e0e68f19d42d7ab diff --git a/lib/Object/Remote/Role/Connector/PerlInterpreter.pm b/lib/Object/Remote/Role/Connector/PerlInterpreter.pm index d418c23..31e13e2 100644 --- a/lib/Object/Remote/Role/Connector/PerlInterpreter.pm +++ b/lib/Object/Remote/Role/Connector/PerlInterpreter.pm @@ -1,18 +1,24 @@ package Object::Remote::Role::Connector::PerlInterpreter; use IPC::Open2; +use IPC::Open3; use IO::Handle; +use Object::Remote::Logging qw( :log :dlog ); use Object::Remote::ModuleSender; use Object::Remote::Handle; use Object::Remote::Future; -use Object::Remote::Logging qw( :log ); -use Scalar::Util qw(blessed); +use Scalar::Util qw(blessed weaken); use Moo::Role; +use Symbol; with 'Object::Remote::Role::Connector'; has module_sender => (is => 'lazy'); +#if no child_stderr file handle is specified then stderr +#of the child will be connected to stderr of the parent +has stderr => ( is => 'rw', default => sub { undef } ); + sub _build_module_sender { my ($hook) = grep {blessed($_) && $_->isa('Object::Remote::ModuleLoader::Hook') } @@ -21,8 +27,16 @@ sub _build_module_sender { } has perl_command => (is => 'lazy'); +has watchdog_timeout => ( is => 'ro', required => 1, default => sub { 0 } ); -sub _build_perl_command { [ 'perl', '-' ] } +#TODO convert nice value into optional feature enabled by +#setting value of attribute +#ulimit of ~500 megs of v-ram +#TODO only works with ssh with quotes but only works locally +#with out quotes +sub _build_perl_command {[ 'sh -c "ulimit -v 200000; nice -n 15 perl -"' ] } +#sub _build_perl_command { [ 'perl', '-' ] } +#sub _build_perl_command { [ 'cat' ] } around connect => sub { my ($orig, $self) = (shift, shift); @@ -30,6 +44,7 @@ around connect => sub { return future { $f->on_done(sub { my ($conn) = $f->get; + $self->_setup_watchdog_reset($conn); my $sub = $conn->remote_sub('Object::Remote::Logging::init_logging_forwarding'); $sub->('Object::Remote::Logging', Object::Remote::Logging->arg_router); Object::Remote::Handle->new( @@ -48,11 +63,57 @@ sub final_perl_command { shift->perl_command } sub _start_perl { my $self = shift; - my $pid = open2( - my $foreign_stdout, + my $given_stderr = $self->stderr; + my $foreign_stderr; + + Dlog_debug { "invoking connection to perl interpreter using command line: $_" } @{$self->final_perl_command}; + + if (defined($given_stderr)) { + #if the stderr data goes to an existing file handle + #an need an anonymous file handle is required + #as the other half of a pipe style file handle pair + #so the file handles can go into the run loop + $foreign_stderr = gensym(); + } else { + #if no file handle has been specified + #for the child's stderr then connect + #the child stderr to the parent stderr + $foreign_stderr = ">&STDERR"; + } + + my $pid = open3( my $foreign_stdin, + my $foreign_stdout, + $foreign_stderr, @{$self->final_perl_command}, ) or die "Failed to run perl at '$_[0]': $!"; + + if (defined($given_stderr)) { + log_warn { "using experimental cat for child stderr" }; + + #TODO refactor if this solves the problem + Object::Remote->current_loop + ->watch_io( + handle => $foreign_stderr, + on_read_ready => sub { + my $buf = ''; + my $len = sysread($foreign_stderr, $buf, 32768); + if (!defined($len) or $len == 0) { + log_trace { "Got EOF or error on child stderr, removing from watcher" }; + $self->stderr(undef); + Object::Remote->current_loop + ->unwatch_io( + handle => $foreign_stderr, + on_read_ready => 1 + ); + } else { + Dlog_trace { "got $len characters of stderr data for connection" }; + print $given_stderr $buf or die "could not send stderr data: $!"; + } + } + ); + } + return ($foreign_stdin, $foreign_stdout, $pid); } @@ -60,32 +121,72 @@ sub _open2_for { my $self = shift; my ($foreign_stdin, $foreign_stdout, $pid) = $self->_start_perl(@_); my $to_send = $self->fatnode_text; + log_debug { my $len = length($to_send); "Sending contents of fat node to remote node; size is '$len' characters" }; Object::Remote->current_loop ->watch_io( handle => $foreign_stdin, on_write_ready => sub { - my $len = syswrite($foreign_stdin, $to_send, 4096); + my $len = syswrite($foreign_stdin, $to_send, 32768); if (defined $len) { substr($to_send, 0, $len) = ''; } # if the stdin went away, we'll never get Shere # so it's not a big deal to simply give up on !defined if (!defined($len) or 0 == length($to_send)) { + log_trace { "Got EOF or error when writing fatnode data to filehandle, unwatching it" }; Object::Remote->current_loop ->unwatch_io( handle => $foreign_stdin, on_write_ready => 1 ); + } else { + log_trace { "Sent $len bytes of fatnode data to remote side" }; } } ); return ($foreign_stdin, $foreign_stdout, $pid); } +sub _setup_watchdog_reset { + my ($self, $conn) = @_; + my $timer_id; + + return unless $self->watchdog_timeout; + + Dlog_trace { "Creating Watchdog management timer for connection id $_" } $conn->_id; + + $timer_id = Object::Remote->current_loop->watch_time( + every => $self->watchdog_timeout / 3, + code => sub { + unless(defined($conn)) { + log_trace { "Weak reference to connection in Watchdog was lost, terminating update timer $timer_id" }; + Object::Remote->current_loop->unwatch_time($timer_id); + return; + } + + Dlog_debug { "Reseting Watchdog for connection id $_" } $conn->_id; + #we do not want to block in the run loop so send the + #update off and ignore any result, we don't need it + #anyway + $conn->send_class_call(0, 'Object::Remote::WatchDog', 'reset'); + } + ); + + $conn->on_close->on_done(sub { Object::Remote->current_loop->unwatch_time($timer_id) }); +} + sub fatnode_text { my ($self) = @_; - require Object::Remote::FatNode; my $text = ''; + + require Object::Remote::FatNode; + + $text = "my \$WATCHDOG_TIMEOUT = '" . $self->watchdog_timeout . "';\n"; + + if (my $duration = $self->watchdog_timeout) { + $text .= "alarm(\$WATCHDOG_TIMEOUT);\n"; + } + $text .= 'BEGIN { $ENV{OBJECT_REMOTE_DEBUG} = 1 }'."\n" if $ENV{OBJECT_REMOTE_DEBUG}; $text .= <<'END'; @@ -98,6 +199,7 @@ END eval $Object::Remote::FatNode::DATA; die $@ if $@; END + $text .= "__END__\n"; return $text; }