X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?p=scpubgit%2FObject-Remote.git;a=blobdiff_plain;f=lib%2FObject%2FRemote%2FConnection.pm;h=136b065aac3284ed345cbe598a5e595b1ac155e2;hp=4fc73055da2fb70c2dd10d754a487878be62f031;hb=b0ec7e3b19d47b9ba5a864f5077d0dc8030834cc;hpb=b1f10b7944d395e404beb995813271d857d143f7 diff --git a/lib/Object/Remote/Connection.pm b/lib/Object/Remote/Connection.pm index 4fc7305..136b065 100644 --- a/lib/Object/Remote/Connection.pm +++ b/lib/Object/Remote/Connection.pm @@ -1,6 +1,6 @@ package Object::Remote::Connection; -use Object::Remote::Logging qw (:log :dlog); +use Object::Remote::Logging qw (:log :dlog router); use Object::Remote::Future; use Object::Remote::Null; use Object::Remote::Handle; @@ -16,37 +16,9 @@ use Module::Runtime qw(use_module); use Scalar::Util qw(weaken blessed refaddr openhandle); use JSON::PP qw(encode_json); use Moo; +use Carp qw(croak); -BEGIN { - #this will reap child processes as soon - #as they are done executing so the process - #table cleans up as fast as possible but - #anything that needs to call waitpid() - #in the future to get the exit value of - #a child will get trash results if - #the signal handler was running. - #If creating a child and getting the - #exit value is required then set - #a localized version of the signal - #handler for CHLD to be 'IGNORE' - #in the smallest block possible - #and outside the block send - #the process a CHLD signal - #to reap anything that may - #have exited while blocked - #in waitpid() - $SIG{CHLD} = sub { - my $kid; - log_trace { "CHLD signal handler is executing" }; - do { - $kid = waitpid(-1, WNOHANG); - log_debug { "waitpid() returned '$kid'" }; - } while $kid > 0; - log_trace { "CHLD signal handler is done" }; - }; - - $SIG{PIPE} = sub { log_debug { "Got a PIPE signal" } }; -} +BEGIN { router()->exclude_forwarding } END { log_debug { "Killing all child processes in the process group" }; @@ -60,9 +32,9 @@ has _id => ( is => 'ro', required => 1, default => sub { our $NEXT_CONNECTION_ID has send_to_fh => ( is => 'ro', required => 1, trigger => sub { - my $self = $_[0]; - $_[1]->autoflush(1); - Dlog_trace { my $id = $self->_id; "connection had send_to_fh set to $_" } $_[1]; + my $self = $_[0]; + $_[1]->autoflush(1); + Dlog_trace { my $id = $self->_id; "connection had send_to_fh set to $_" } $_[1]; }, ); @@ -71,11 +43,11 @@ has read_channel => ( trigger => sub { my ($self, $ch) = @_; my $id = $self->_id; - Dlog_trace { "trigger for read_channel has been invoked for connection $id; file handle is $_" } $ch->fh; + Dlog_trace { "trigger for read_channel has been invoked for connection $id; file handle is $_" } $ch->fh; weaken($self); $ch->on_line_call(sub { $self->_receive(@_) }); $ch->on_close_call(sub { - log_trace { "invoking 'done' on on_close handler for connection id '$id'" }; + log_trace { "invoking 'done' on on_close handler for connection id '$id'" }; $self->on_close->done(@_); }); }, @@ -83,7 +55,10 @@ has read_channel => ( has on_close => ( is => 'rw', default => sub { $_[0]->_install_future_handlers(CPS::Future->new) }, - trigger => \&_install_future_handlers, + trigger => sub { + log_trace { "Installing handlers into future via trigger" }; + $_[0]->_install_future_handlers($_[1]) + }, ); has child_pid => (is => 'ro'); @@ -109,33 +84,69 @@ has _json => ( ); after BUILD => sub { - my ($self) = @_; - - return unless defined $self->child_pid; + my ($self) = @_; + my $pid = $self->child_pid; - log_debug { "Setting process group of child process" }; + unless (defined $pid) { + log_trace { "After BUILD invoked for connection but there was no pid" }; + return; + } + + log_trace { "Setting process group of child process '$pid'" }; setpgrp($self->child_pid, 1); }; sub BUILD { } +sub is_valid { + my ($self) = @_; + my $closed = $self->on_close->is_ready; + + log_trace { "Connection closed: $closed" }; + return ! $closed; +} + sub _fail_outstanding { my ($self, $error) = @_; - Dlog_debug { "Failing outstanding futures with '$error' for connection $_" } $self->_id; my $outstanding = $self->outstanding_futures; - $_->fail("$error\n") for values %$outstanding; + + Dlog_debug { + sprintf "Failing %i outstanding futures with '$error'", scalar(keys(%$outstanding)) + }; + + foreach(keys(%$outstanding)) { + log_trace { "Failing future for $_" }; + my $future = $outstanding->{$_}; + $future->fail("$error\n"); + } + %$outstanding = (); return; } sub _install_future_handlers { my ($self, $f) = @_; - Dlog_trace { "trigger for on_close has been invoked for connection $_" } $self->_id; + Dlog_trace { "Installing handlers into future for connection $_" } $self->_id; weaken($self); $f->on_done(sub { - Dlog_trace { "failing all of the outstanding futures for connection $_" } $self->_id; + my $pid = $self->child_pid; + Dlog_trace { "Executing on_done handler in future for connection $_" } $self->_id; $self->_fail_outstanding("Object::Remote connection lost: " . ($f->get)[0]); + return unless defined $pid; + log_debug { "Waiting for child '$pid' to exit" }; + my $ret = waitpid($pid, 0); + if ($ret != $pid) { + log_debug { "Waited for pid $pid but waitpid() returned $ret" }; + return; + } elsif ($? & 127) { + log_warn { "Remote interpreter did not exit cleanly" }; + } else { + log_verbose { + my $exit_value = $? >> 8; + "Remote Perl interpreter exited with value '$exit_value'" + }; + } }); return $f; }; @@ -225,9 +236,9 @@ sub conn_from_spec { } sub new_from_spec { - my ($class, $spec) = @_; + my ($class, $spec, @args) = @_; return $spec if blessed $spec; - my $conn = $class->conn_from_spec($spec); + my $conn = $class->conn_from_spec($spec, @args); die "Couldn't figure out what to do with ${spec}" unless defined $conn; @@ -291,6 +302,9 @@ sub register_remote { sub send_free { my ($self, $id) = @_; Dlog_trace { "sending request to free object '$id' for connection $_" } $self->_id; + #TODO this shows up some times when a remote side dies in the middle of a remote + #method invocation - possibly only when the object is being constructed? + #(in cleanup) Use of uninitialized value $id in delete at ../Object-Remote/lib/Object/Remote/Connection. delete $self->remote_objects_by_id->{$id}; $self->_send([ free => $id ]); } @@ -326,6 +340,11 @@ sub send_discard { sub _send { my ($self, $to_send) = @_; my $fh = $self->send_to_fh; + + unless ($self->is_valid) { + croak "Attempt to invoke _send on a connection that is not valid"; + } + Dlog_trace { "Starting to serialize data in argument to _send for connection $_" } $self->_id; my $serialized = $self->_serialize($to_send)."\n"; Dlog_trace { my $l = length($serialized); "serialization is completed; sending '$l' characters of serialized data to $_" } $fh; @@ -340,7 +359,8 @@ sub _send { if ($@) { Dlog_debug { "exception encountered when trying to write to file handle $_: $@" } $fh; - my $error = $@; chomp($error); + my $error = $@; + chomp($error); $self->on_close->done("could not write to file handle: $error") unless $self->on_close->is_ready; return; } @@ -480,8 +500,82 @@ sub _invoke { Object::Remote::Connection - An underlying connection for L -=head1 LAME + use Object::Remote; + + my %opts = ( + nice => '10', ulimit => '-v 400000', + watchdog_timeout => 120, stderr => \*STDERR, + ); + + my $local = Object::Remote->connect('-'); + my $remote = Object::Remote->connect('myserver', nice => 5); + my $remote_user = Object::Remote->connect('user@myserver', %opts); + my $local_sudo = Object::Remote->connect('user@'); + + #$remote can be any other connection object + my $hostname = Sys::Hostname->can::on($remote, 'hostname'); + +=head1 DESCRIPTION + +This is the class that supports connections to a Perl interpreter that is executed in a +different process. The new Perl interpreter can be either on the local or a remote machine +and is configurable via arguments passed to the constructor. + +=head1 ARGUMENTS + +=over 4 + +=item nice + +If this value is defined then it will be used as the nice value of the Perl process when it +is started. The default is the undefined value and will not nice the process. + +=item stderr + +If this value is defined then it will be used as the file handle that receives the output +of STDERR from the Perl interpreter process and I/O will be performed by the run loop in a +non-blocking way. If the value is undefined then STDERR of the remote process will be connected +directly to STDERR of the local process with out the run loop managing I/O. The default value +is undefined. + +There are a few ways to use this feature. By default the behavior is to form one unified STDERR +across all of the Perl interpreters including the local one. For small scale and quick operation +this offers a predictable and easy to use way to get at error messages generated anywhere. If +the local Perl interpreter crashes then the remote Perl interpreters still have an active STDERR +and it is possible to still receive output from them. This is generally a good thing but can +cause issues. + +When using a file handle as the output for STDERR once the local Perl interpreter is no longer +running there is no longer a valid STDERR for the remote interpreters to send data to. This means +that it is no longer possible to receive error output from the remote interpreters and that the +shell will start to kill off the child processes. Passing a reference to STDERR for the local +interpreter (as the SYNOPSIS shows) causes the run loop to manage I/O, one unified STDERR for +all Perl interpreters that ends as soon as the local interpreter process does, and the shell will +start killing children when the local interpreter exits. + +It is also possible to pass in a file handle that has been opened for writing. This would be +useful for logging the output of the remote interpreter directly into a dedicated file. + +=item ulimit + +If this string is defined then it will be passed unmodified as the arguments to ulimit when +the Perl process is started. The default value is the undefined value and will not limit the +process in any way. + +=item watchdog_timeout + +If this value is defined then it will be used as the number of seconds the watchdog will wait +for an update before it terminates the Perl interpreter process. The default value is undefined +and will not use the watchdog. See C for more information. + +=back + +=head1 SEE ALSO + +=over 4 + +=item C -Shipping prioritised over writing this part up. Blame mst. +=back =cut