make notes of things to fix before release

[scpubgit/Object-Remote.git] / lib / Object / Remote / Connection.pm

package Object::Remote::Connection;

use Object::Remote::Logging qw (:log :dlog router);
use Object::Remote::Future;
use Object::Remote::Null;
use Object::Remote::Handle;
use Object::Remote::CodeContainer;
use Object::Remote::GlobProxy;
use Object::Remote::GlobContainer;
use Object::Remote::Tied;
use Object::Remote;
use Symbol;
use IO::Handle;
use POSIX ":sys_wait_h";
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 { router()->exclude_forwarding }

END {
  log_debug { "Killing all child processes in the process group" };
    
  #FIXME update along with setpgrp() to not use a process
  #group anymore
  
  #send SIGINT to the process group for our children
  kill(1, -2);
}

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];
  },
);

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;
    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(@_);
    });
  },
);

has on_close => (
  is => 'rw', default => sub { $_[0]->_install_future_handlers(CPS::Future->new) },
  trigger => sub {
    log_trace { "Installing handlers into future via trigger" };
    $_[0]->_install_future_handlers($_[1])
  },
);

has child_pid => (is => 'ro');

has local_objects_by_id => (
  is => 'ro', default => sub { {} },
  coerce => sub { +{ %{$_[0]} } }, # shallow clone on the way in
);

has remote_objects_by_id => (
  is => 'ro', default => sub { {} },
  coerce => sub { +{ %{$_[0]} } }, # shallow clone on the way in
);

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'" };
  
  #FIXME moving things into a process group has side effects for
  #users of the library - move to a list
  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) = @_;
  my $outstanding = $self->outstanding_futures;
  
  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 { "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) = @_;
  Dlog_trace { "fetching proxy for remote object with id '$id' for connection $_" } $self->_id;
  return bless({}, 'Object::Remote::Null') if $id eq 'NULL';
  (
    $self->remote_objects_by_id->{$id}
    or Object::Remote::Handle->new(connection => $self, id => $id)
  )->proxy;
}

sub _build__json {
  weaken(my $self = shift);
  JSON::PP->new->filter_json_single_key_object(
    __remote_object__ => sub {
      $self->_id_to_remote_object(@_);
    }
  )->filter_json_single_key_object(
    __remote_code__ => sub {
      my $code_container = $self->_id_to_remote_object(@_);
      sub { $code_container->call(@_) };
    }
  )->filter_json_single_key_object(
    __scalar_ref__ => sub {
      my $value = shift;
      return \$value;
    }
  )->filter_json_single_key_object(
    __glob_ref__ => sub {
      my $glob_container = $self->_id_to_remote_object(@_);
      my $handle = Symbol::gensym;
      tie *$handle, 'Object::Remote::GlobProxy', $glob_container;
      return $handle;
    }
  )->filter_json_single_key_object(
    __local_object__ => sub {
      $self->local_objects_by_id->{$_[0]}
    }
  )->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;
    }
  )->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;
    }
  ); 
}

sub _load_if_possible {
  my ($class) = @_; 

  use_module($class); 

  if ($@) {
    log_debug { "Attempt at loading '$class' failed with '$@'" };
  }

}

BEGIN {
  unshift our @Guess, sub { blessed($_[0]) ? $_[0] : undef };
  map _load_if_possible($_), qw(
    Object::Remote::Connector::Local
    Object::Remote::Connector::LocalSudo
    Object::Remote::Connector::SSH
    Object::Remote::Connector::UNIX
  ); 
}

sub conn_from_spec {
  my ($class, $spec, @args) = @_;
  foreach my $poss (do { our @Guess }) {
    if (my $conn = $poss->($spec, @args)) {
      return $conn;
    }
  }
  
  return undef;
}

sub new_from_spec {
  my ($class, $spec, @args) = @_;
  return $spec if blessed $spec;
  my $conn = $class->conn_from_spec($spec, @args);
  
  die "Couldn't figure out what to do with ${spec}"
    unless defined $conn;
    
  return $conn->maybe::start::connect;  
}

sub remote_object {
  my ($self, @args) = @_;
  Object::Remote::Handle->new(
    connection => $self, @args
  )->proxy;
}

sub connect {
  my ($self, $to) = @_;
  Dlog_debug { "Creating connection to remote node '$to' for connection $_" } $self->_id;
  return await_future(
    $self->send_class_call(0, 'Object::Remote', connect => $to)
  );
}

sub remote_sub {
  my ($self, $sub) = @_;
  my ($pkg, $name) = $sub =~ m/^(.*)::([^:]+)$/;
  Dlog_debug { "Invoking remote sub '$sub' for connection $_" } $self->_id;
  return await_future($self->send_class_call(0, $pkg, can => $name));
}

sub send_class_call {
  my ($self, $ctx, @call) = @_;
  Dlog_trace { "Sending a class call for connection $_" } $self->_id;
  $self->send(call => class_call_handler => $ctx => call => @call);
}

sub register_class_call_handler {
  my ($self) = @_;
  $self->local_objects_by_id->{'class_call_handler'} ||= do {
    my $o = $self->new_class_call_handler;
    $self->_local_object_to_id($o);
    $o;
  };
}

sub new_class_call_handler {
  Object::Remote::CodeContainer->new(
    code => sub {
      my ($class, $method) = (shift, shift);
      use_module($class)->$method(@_);
    }
  );
}

sub register_remote {
  my ($self, $remote) = @_;
  Dlog_trace { my $i = $remote->id; "Registered a remote object with id of '$i' for connection $_" } $self->_id;
  weaken($self->remote_objects_by_id->{$remote->id} = $remote);
  return $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 ]);
}

sub send {
  my ($self, $type, @call) = @_;

  my $future = CPS::Future->new;
  my $remote = $self->remote_objects_by_id->{$call[0]};

  unshift @call, $type => $self->_local_object_to_id($future);

  my $outstanding = $self->outstanding_futures;
  $outstanding->{$future} = $future;
  $future->on_ready(sub {
    undef($remote);
    delete $outstanding->{$future}
  });

  $self->_send(\@call);

  return $future;
}

sub send_discard {
  my ($self, $type, @call) = @_;

  unshift @call, $type => 'NULL';

  $self->_send(\@call);
}

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;
  my $ret; 
  eval { 
    #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; 
  }
      
  return $ret; 
}

sub _serialize {
  my ($self, $data) = @_;
  local our @New_Ids = (-1);
  return eval {
    my $flat = $self->_encode($self->_deobjectify($data));
    $flat;
  } || do {
    my $err = $@; # won't get here if the eval doesn't die
    # don't keep refs to new things
    delete @{$self->local_objects_by_id}{@New_Ids};
    die "Error serializing: $err";
  };
}

sub _local_object_to_id {
  my ($self, $object) = @_;
  my $id = refaddr($object);
  $self->local_objects_by_id->{$id} ||= do {
    push our(@New_Ids), $id if @New_Ids;
    $object;
  };
  return $id;
}

sub _deobjectify {
  my ($self, $data) = @_;
  if (blessed($data)) {
    if (
      $data->isa('Object::Remote::Proxy')
      and $data->{remote}->connection == $self
    ) {
      return +{ __local_object__ => $data->{remote}->id };
    } else {
      return +{ __remote_object__ => $self->_local_object_to_id($data) };
    }
  } elsif (my $ref = ref($data)) {
    if ($ref eq 'HASH') {
      my $tied_to = tied(%$data);
      if(defined($tied_to)) {
        return +{__remote_tied_hash__ => $self->_local_object_to_id($tied_to)}; 
      } else {
        return +{ map +($_ => $self->_deobjectify($data->{$_})), keys %$data };
      }
    } elsif ($ref eq 'ARRAY') {
      my $tied_to = tied(@$data);
      if (defined($tied_to)) {
        return +{__remote_tied_array__ => $self->_local_object_to_id($tied_to)}; 
      } else {
        return [ map $self->_deobjectify($_), @$data ];
      }
    } elsif ($ref eq 'CODE') {
      my $id = $self->_local_object_to_id(
                 Object::Remote::CodeContainer->new(code => $data)
               );
      return +{ __remote_code__ => $id };
    } elsif ($ref eq 'SCALAR') {
      return +{ __scalar_ref__ => $$data };
    } elsif ($ref eq 'GLOB') {
      return +{ __glob_ref__ => $self->_local_object_to_id(
        Object::Remote::GlobContainer->new(handle => $data)
      ) };
    } else {
      die "Can't collapse reftype $ref";
    }
  }
  return $data; # plain scalar
}

sub _receive {
  my ($self, $flat) = @_;
  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 };
  Dlog_trace { "deserialization complete for connection $_" } $self->_id;
  eval { $self->${\"receive_${type}"}(@rest); 1 }
    or do { warn "Receive failed for ${flat}: $@"; return };
  return;
}

sub receive_free {
  my ($self, $id) = @_;
  Dlog_trace { "got a receive_free for object '$id' for connection $_" } $self->_id;
  delete $self->local_objects_by_id->{$id}
    or warn "Free: no such object $id";
  return;
}

sub receive_call {
  my ($self, $future_id, $id, @rest) = @_;
  Dlog_trace { "got a receive_call for object '$id' for connection $_" } $self->_id;
  my $future = $self->_id_to_remote_object($future_id);
  $future->{method} = 'call_discard_free';
  my $local = $self->local_objects_by_id->{$id}
    or do { $future->fail("No such object $id"); return };
  $self->_invoke($future, $local, @rest);
}

sub receive_call_free {
  my ($self, $future, $id, @rest) = @_;
  Dlog_trace { "got a receive_call_free for object '$id' for connection $_" } $self->_id;
  $self->receive_call($future, $id, undef, @rest);
  $self->receive_free($id);
}

sub _invoke {
  my ($self, $future, $local, $ctx, $method, @args) = @_;
  Dlog_trace { "got _invoke for a method named '$method' for connection $_" } $self->_id;
  if ($method =~ /^start::/) {
    my $f = $local->$method(@args);
    $f->on_done(sub { undef($f); $future->done(@_) });
    return unless $f;
    $f->on_fail(sub { undef($f); $future->fail(@_) });
    return;
  }
  my $do = sub { $local->$method(@args) };
  eval {
    $future->done(
      defined($ctx)
        ? ($ctx ? $do->() : scalar($do->()))
        : do { $do->(); () }
    );
    1;
  } or do { $future->fail($@); return; };
  return;
}

1;

=head1 NAME

Object::Remote::Connection - An underlying connection for L<Object::Remote>

  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<Object::Remote::Watchdog> for more information.

=back

=head1 SEE ALSO

=over 4

=item C<Object::Remote>

=back

=cut