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=3832bfab0b3054b8579ccda7f7e3e019d267eee8;hp=3f1574877036a0872a6b3baeefb023f40c704902;hb=302ecfbf4056a977f44f185cbfa925e3886d0c1a;hpb=b7a853b30e7397854a813f6a5878e50703a20079 diff --git a/lib/Object/Remote/Connection.pm b/lib/Object/Remote/Connection.pm index 3f15748..3832bfa 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; @@ -18,34 +18,8 @@ use JSON::PP qw(encode_json); use Moo; 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" } }; + router()->exclude_forwarding; + $SIG{PIPE} = sub { log_debug { "Got a PIPE signal" } }; } END { @@ -55,12 +29,6 @@ END { kill(1, -2); } - -our $DEBUG = !!$ENV{OBJECT_REMOTE_DEBUG}; -#numbering each connection allows it to be -#tracked along with file handles in -#the logs - has _id => ( is => 'ro', required => 1, default => sub { our $NEXT_CONNECTION_ID++ } ); has send_to_fh => ( @@ -72,56 +40,27 @@ has send_to_fh => ( }, ); -#TODO see if this is another case of the same bug below -#where trigger never fires because the attribute isn't -#actually set at any time has read_channel => ( is => 'ro', required => 1, 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'" }; - $self->on_close->done(@_); + log_trace { "invoking 'done' on on_close handler for connection id '$id'" }; + $self->on_close->done(@_); }); }, ); -#TODO properly fix this bug - -#trigger can't ever be invoked with a default -#value and the on_close attribute is read only.... -#the future never gets the on_done handler -#installed -sub BUILD { - my ($self) = @_; - $self->on_close(CPS::Future->new); -} - -after BUILD => sub { - my ($self) = @_; - - return unless defined $self->child_pid; - - log_debug { "Setting process group of child process" }; - - setpgrp($self->child_pid, 1); -}; - - has on_close => ( - is => 'rw', default => sub { CPS::Future->new }, + is => 'rw', default => sub { $_[0]->_install_future_handlers(CPS::Future->new) }, trigger => sub { - my ($self, $f) = @_; - Dlog_trace { "trigger for on_close has been invoked for connection $_" } $self->_id; - weaken($self); - $f->on_done(sub { - Dlog_trace { "failing all of the outstanding futures for connection $_" } $self->_id; - $self->_fail_outstanding("Object::Remote connection lost: " . ($f->get)[0]); - }); - } + log_trace { "Installing handlers into future via trigger" }; + $_[0]->_install_future_handlers($_[1]) + }, ); has child_pid => (is => 'ro'); @@ -138,22 +77,64 @@ has remote_objects_by_id => ( has outstanding_futures => (is => 'ro', default => sub { {} }); +has _json => ( + is => 'lazy', + handles => { + _deserialize => 'decode', + _encode => 'encode', + }, +); + +after BUILD => sub { + my ($self) = @_; + my $pid = $self->child_pid; + + 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 _fail_outstanding { my ($self, $error) = @_; - Dlog_debug { "Failing outstanding futures with '$error' for connection $_" } $self->_id; + Dlog_debug { "$$ Failing outstanding futures with '$error' for connection $_" } $self->_id; my $outstanding = $self->outstanding_futures; $_->fail("$error\n") for values %$outstanding; %$outstanding = (); return; } -has _json => ( - is => 'lazy', - handles => { - _deserialize => 'decode', - _encode => 'encode', - }, -); +sub _install_future_handlers { + my ($self, $f) = @_; + Dlog_trace { "Installing handlers into future for connection $_" } $self->_id; + weaken($self); + $f->on_done(sub { + 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; +}; sub _id_to_remote_object { my ($self, $id) = @_; @@ -194,15 +175,15 @@ sub _build__json { } )->filter_json_single_key_object( __remote_tied_hash__ => sub { - my %tied_hash; - tie %tied_hash, 'Object::Remote::Tied', $self->_id_to_remote_object(@_); - return \%tied_hash; + my %tied_hash; + tie %tied_hash, 'Object::Remote::Tied', $self->_id_to_remote_object(@_); + return \%tied_hash; } )->filter_json_single_key_object( __remote_tied_array__ => sub { - my @tied_array; - tie @tied_array, 'Object::Remote::Tied', $self->_id_to_remote_object(@_); - return \@tied_array; + my @tied_array; + tie @tied_array, 'Object::Remote::Tied', $self->_id_to_remote_object(@_); + return \@tied_array; } ); } @@ -210,7 +191,7 @@ sub _build__json { sub _load_if_possible { my ($class) = @_; - eval "require $class"; + use_module($class); if ($@) { log_debug { "Attempt at loading '$class' failed with '$@'" }; @@ -306,6 +287,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 ]); } @@ -344,24 +328,21 @@ sub _send { 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; - #TODO this is very risky for deadlocks unless it's set to non-blocking and then with out extra - #logic it could easily do short-writes to the remote side - how about taking this entire buffer - #and having the run loop send it to the file handle so this doesn't block while the sending - #is happening? my $ret; eval { - local($SIG{PIPE}) = 'IGNORE'; - die "filehandle is not open" unless openhandle($fh); - log_trace { "file handle has passed openhandle() test; printing to it" }; - $ret = print $fh $serialized; - die "print was not successful: $!" unless defined $ret + #TODO this should be converted over to a non-blocking ::WriteChannel class + die "filehandle is not open" unless openhandle($fh); + log_trace { "file handle has passed openhandle() test; printing to it" }; + $ret = print $fh $serialized; + die "print was not successful: $!" unless defined $ret }; if ($@) { - Dlog_debug { "exception encountered when trying to write to file handle $_: $@" } $fh; - my $error = $@; chomp($error); - $self->on_close->done("could not write to file handle: $error") unless $self->on_close->is_ready; - return; + Dlog_debug { "exception encountered when trying to write to file handle $_: $@" } $fh; + my $error = $@; + chomp($error); + $self->on_close->done("could not write to file handle: $error") unless $self->on_close->is_ready; + return; } return $ret; @@ -372,7 +353,6 @@ sub _serialize { local our @New_Ids = (-1); return eval { my $flat = $self->_encode($self->_deobjectify($data)); - warn "$$ >>> ${flat}\n" if $DEBUG; $flat; } || do { my $err = $@; # won't get here if the eval doesn't die @@ -438,7 +418,6 @@ sub _deobjectify { sub _receive { my ($self, $flat) = @_; - warn "$$ <<< $flat\n" if $DEBUG; Dlog_trace { my $l = length($flat); "Starting to deserialize $l characters of data for connection $_" } $self->_id; my ($type, @rest) = eval { @{$self->_deserialize($flat)} } or do { warn "Deserialize failed for ${flat}: $@"; return };