lib/Object/Remote/Connection.pm

   1 package Object::Remote::Connection;
   2
   3 use Object::Remote::Future;
   4 use Object::Remote::Null;
   5 use Object::Remote::Handle;
   6 use Object::Remote::CodeContainer;
   7 use Object::Remote::GlobProxy;
   8 use Object::Remote::GlobContainer;
   9 use Object::Remote::Logging qw (:log :dlog);
  10 use Object::Remote::Tied;
  11 use Object::Remote;
  12 use Symbol;
  13 use IO::Handle;
  14 use Module::Runtime qw(use_module);
  15 use Scalar::Util qw(weaken blessed refaddr);
  16 use JSON::PP qw(encode_json);
  17 use Moo;
  18
  19 our $DEBUG = !!$ENV{OBJECT_REMOTE_DEBUG};
  20 #numbering each connection allows it to be
  21 #tracked along with file handles in
  22 #the logs
  23 BEGIN { our $NEXT_CONNECTION_ID = 0 }
  24 has _id => ( is => 'ro', required => 1, default => sub { our $NEXT_CONNECTION_ID++ } );
  25
  26 has send_to_fh => (
  27   is => 'ro', required => 1,
  28   trigger => sub {
  29       my $self = $_[0];
  30       $_[1]->autoflush(1);
  31       Dlog_trace { my $id = $self->_id; "connection had send_to_fh set to $_"  } $_[1];
  32   },
  33 );
  34
  35 has read_channel => (
  36   is => 'ro', required => 1,
  37   trigger => sub {
  38     my ($self, $ch) = @_;
  39     Dlog_trace { my $id = $self->_id; "trigger for read_channel has been invoked for connection $id; file handle is " } $ch->fh;
  40     weaken($self);
  41     $ch->on_line_call(sub { $self->_receive(@_) });
  42     $ch->on_close_call(sub { $self->on_close->done(@_) });
  43   },
  44 );
  45
  46 has on_close => (
  47   is => 'ro', default => sub { CPS::Future->new },
  48   trigger => sub {
  49     my ($self, $f) = @_;
  50     Dlog_trace { "trigger for on_close has been invoked for connection $_" } $self->_id;
  51     weaken($self);
  52     $f->on_done(sub {
  53       $self->_fail_outstanding("Connection lost: ".($f->get)[0]);
  54     });
  55   }
  56 );
  57
  58 has child_pid => (is => 'ro');
  59
  60 has local_objects_by_id => (
  61   is => 'ro', default => sub { {} },
  62   coerce => sub { +{ %{$_[0]} } }, # shallow clone on the way in
  63 );
  64
  65 has remote_objects_by_id => (
  66   is => 'ro', default => sub { {} },
  67   coerce => sub { +{ %{$_[0]} } }, # shallow clone on the way in
  68 );
  69
  70 has outstanding_futures => (is => 'ro', default => sub { {} });
  71
  72 sub _fail_outstanding {
  73   my ($self, $error) = @_;
  74   Dlog_debug { "Failing outstanding futures with '$error' for connection $_" } $self->_id;
  75   my $outstanding = $self->outstanding_futures;
  76   $_->fail($error) for values %$outstanding;
  77   %$outstanding = ();
  78   return;
  79 }
  80
  81 has _json => (
  82   is => 'lazy',
  83   handles => {
  84     _deserialize => 'decode',
  85     _encode => 'encode',
  86   },
  87 );
  88
  89 sub _id_to_remote_object {
  90   my ($self, $id) = @_;
  91   Dlog_trace { "fetching proxy for remote object with id '$id' for connection $_" } $self->_id;
  92   return bless({}, 'Object::Remote::Null') if $id eq 'NULL';
  93   (
  94     $self->remote_objects_by_id->{$id}
  95     or Object::Remote::Handle->new(connection => $self, id => $id)
  96   )->proxy;
  97 }
  98
  99 sub _build__json {
 100   weaken(my $self = shift);
 101   JSON::PP->new->filter_json_single_key_object(
 102     __remote_object__ => sub {
 103       $self->_id_to_remote_object(@_);
 104     }
 105   )->filter_json_single_key_object(
 106     __remote_code__ => sub {
 107       my $code_container = $self->_id_to_remote_object(@_);
 108       sub { $code_container->call(@_) };
 109     }
 110   )->filter_json_single_key_object(
 111     __scalar_ref__ => sub {
 112       my $value = shift;
 113       return \$value;
 114     }
 115   )->filter_json_single_key_object(
 116     __glob_ref__ => sub {
 117       my $glob_container = $self->_id_to_remote_object(@_);
 118       my $handle = Symbol::gensym;
 119       tie *$handle, 'Object::Remote::GlobProxy', $glob_container;
 120       return $handle;
 121     }
 122   )->filter_json_single_key_object(
 123     __local_object__ => sub {
 124       $self->local_objects_by_id->{$_[0]}
 125     }
 126   )->filter_json_single_key_object(
 127     __remote_tied_hash__ => sub {
 128         my %tied_hash;
 129         tie %tied_hash, 'Object::Remote::Tied', $self->_id_to_remote_object(@_);
 130         return \%tied_hash;
 131     }
 132   )->filter_json_single_key_object(
 133     __remote_tied_array__ => sub {
 134         my @tied_array;
 135         tie @tied_array, 'Object::Remote::Tied', $self->_id_to_remote_object(@_);
 136         return \@tied_array;
 137     }
 138   );
 139 }
 140
 141 BEGIN {
 142   unshift our @Guess, sub { blessed($_[0]) ? $_[0] : undef };
 143   eval { require Object::Remote::Connector::Local };
 144   eval { require Object::Remote::Connector::LocalSudo };
 145   eval { require Object::Remote::Connector::SSH };
 146   eval { require Object::Remote::Connector::UNIX };
 147 }
 148
 149 sub new_from_spec {
 150   my ($class, $spec) = @_;
 151   return $spec if blessed $spec;
 152   Dlog_debug { "creating a new connection from spec" };
 153   foreach my $poss (do { our @Guess }) {
 154     if (my $conn = $poss->($spec)) {
 155       return $conn->maybe::start::connect;
 156     }
 157   }
 158   die "Couldn't figure out what to do with ${spec}";
 159 }
 160
 161 sub remote_object {
 162   my ($self, @args) = @_;
 163   Object::Remote::Handle->new(
 164     connection => $self, @args
 165   )->proxy;
 166 }
 167
 168 sub connect {
 169   my ($self, $to) = @_;
 170   Dlog_debug { "Creating connection to remote node '$to' for connection $_" } $self->_id;
 171   return await_future(
 172     $self->send_class_call(0, 'Object::Remote', connect => $to)
 173   );
 174 }
 175
 176 sub remote_sub {
 177   my ($self, $sub) = @_;
 178   my ($pkg, $name) = $sub =~ m/^(.*)::([^:]+)$/;
 179   Dlog_debug { "Invoking remote sub '$sub' for connection $_" } $self->_id;
 180   return await_future($self->send_class_call(0, $pkg, can => $name));
 181 }
 182
 183 sub send_class_call {
 184   my ($self, $ctx, @call) = @_;
 185   Dlog_trace { "Sending a class call for connection $_" } $self->_id;
 186   $self->send(call => class_call_handler => $ctx => call => @call);
 187 }
 188
 189 sub register_class_call_handler {
 190   my ($self) = @_;
 191   $self->local_objects_by_id->{'class_call_handler'} ||= do {
 192     my $o = $self->new_class_call_handler;
 193     $self->_local_object_to_id($o);
 194     $o;
 195   };
 196 }
 197
 198 sub new_class_call_handler {
 199   Object::Remote::CodeContainer->new(
 200     code => sub {
 201       my ($class, $method) = (shift, shift);
 202       use_module($class)->$method(@_);
 203     }
 204   );
 205 }
 206
 207 sub register_remote {
 208   my ($self, $remote) = @_;
 209   Dlog_trace { my $i = $remote->id; "Registered a remote object with id of '$i' for connection $_" } $self->_id;
 210   weaken($self->remote_objects_by_id->{$remote->id} = $remote);
 211   return $remote;
 212 }
 213
 214 sub send_free {
 215   my ($self, $id) = @_;
 216   Dlog_trace { "sending request to free object '$id' for connection $_" } $self->_id;
 217   delete $self->remote_objects_by_id->{$id};
 218   $self->_send([ free => $id ]);
 219 }
 220
 221 sub send {
 222   my ($self, $type, @call) = @_;
 223
 224   my $future = CPS::Future->new;
 225   my $remote = $self->remote_objects_by_id->{$call[0]};
 226
 227   unshift @call, $type => $self->_local_object_to_id($future);
 228
 229   my $outstanding = $self->outstanding_futures;
 230   $outstanding->{$future} = $future;
 231   $future->on_ready(sub {
 232     undef($remote);
 233     delete $outstanding->{$future}
 234   });
 235
 236   $self->_send(\@call);
 237
 238   return $future;
 239 }
 240
 241 sub send_discard {
 242   my ($self, $type, @call) = @_;
 243
 244   unshift @call, $type => 'NULL';
 245
 246   $self->_send(\@call);
 247 }
 248
 249 sub _send {
 250   my ($self, $to_send) = @_;
 251   my $fh = $self->send_to_fh;
 252   Dlog_trace { "Starting to serialize data in argument to _send for connection $_" } $self->_id;
 253   my $serialized = $self->_serialize($to_send)."\n";
 254   Dlog_trace { my $l = length($serialized); "serialization is completed; sending '$l' characters of serialized data to $_" } $fh;
 255   #TODO this is very risky for deadlocks unless it's set to non-blocking and then with out extra
 256   #logic it could easily do short-writes to the remote side - how about taking this entire buffer
 257   #and having the run loop send it to the file handle so this doesn't block while the sending
 258   #is happening?
 259   my $ret = print $fh $serialized;
 260   Dlog_trace { my $r = defined $ret ? $ret : 'undef'; "print() returned $r with $_" } $fh;
 261   #TODO hrm reason print's return value was ignored?
 262   die "could not write to filehandle: $!" unless $ret;
 263   return $ret;
 264 }
 265
 266 sub _serialize {
 267   my ($self, $data) = @_;
 268   local our @New_Ids = (-1);
 269   return eval {
 270     my $flat = $self->_encode($self->_deobjectify($data));
 271     warn "$$ >>> ${flat}\n" if $DEBUG;
 272     $flat;
 273   } || do {
 274     my $err = $@; # won't get here if the eval doesn't die
 275     # don't keep refs to new things
 276     delete @{$self->local_objects_by_id}{@New_Ids};
 277     die "Error serializing: $err";
 278   };
 279 }
 280
 281 sub _local_object_to_id {
 282   my ($self, $object) = @_;
 283   my $id = refaddr($object);
 284   $self->local_objects_by_id->{$id} ||= do {
 285     push our(@New_Ids), $id if @New_Ids;
 286     $object;
 287   };
 288   return $id;
 289 }
 290
 291 sub _deobjectify {
 292   my ($self, $data) = @_;
 293   if (blessed($data)) {
 294     if (
 295       $data->isa('Object::Remote::Proxy')
 296       and $data->{remote}->connection == $self
 297     ) {
 298       return +{ __local_object__ => $data->{remote}->id };
 299     } else {
 300       return +{ __remote_object__ => $self->_local_object_to_id($data) };
 301     }
 302   } elsif (my $ref = ref($data)) {
 303     if ($ref eq 'HASH') {
 304       my $tied_to = tied(%$data);
 305       if(defined($tied_to)) {
 306         return +{__remote_tied_hash__ => $self->_local_object_to_id($tied_to)};
 307       } else {
 308         return +{ map +($_ => $self->_deobjectify($data->{$_})), keys %$data };
 309       }
 310     } elsif ($ref eq 'ARRAY') {
 311       my $tied_to = tied(@$data);
 312       if (defined($tied_to)) {
 313         return +{__remote_tied_array__ => $self->_local_object_to_id($tied_to)};
 314       } else {
 315         return [ map $self->_deobjectify($_), @$data ];
 316       }
 317     } elsif ($ref eq 'CODE') {
 318       my $id = $self->_local_object_to_id(
 319                  Object::Remote::CodeContainer->new(code => $data)
 320                );
 321       return +{ __remote_code__ => $id };
 322     } elsif ($ref eq 'SCALAR') {
 323       return +{ __scalar_ref__ => $$data };
 324     } elsif ($ref eq 'GLOB') {
 325       return +{ __glob_ref__ => $self->_local_object_to_id(
 326         Object::Remote::GlobContainer->new(handle => $data)
 327       ) };
 328     } else {
 329       die "Can't collapse reftype $ref";
 330     }
 331   }
 332   return $data; # plain scalar
 333 }
 334
 335 sub _receive {
 336   my ($self, $flat) = @_;
 337   warn "$$ <<< $flat\n" if $DEBUG;
 338   Dlog_trace { my $l = length($flat); "Starting to deserialize $l characters of data for connection $_" } $self->_id;
 339   my ($type, @rest) = eval { @{$self->_deserialize($flat)} }
 340     or do { warn "Deserialize failed for ${flat}: $@"; return };
 341   Dlog_trace { "deserialization complete for connection $_" } $self->_id;
 342   eval { $self->${\"receive_${type}"}(@rest); 1 }
 343     or do { warn "Receive failed for ${flat}: $@"; return };
 344   return;
 345 }
 346
 347 sub receive_free {
 348   my ($self, $id) = @_;
 349   Dlog_trace { "got a receive_free for object '$id' for connection $_" } $self->_id;
 350   delete $self->local_objects_by_id->{$id}
 351     or warn "Free: no such object $id";
 352   return;
 353 }
 354
 355 sub receive_call {
 356   my ($self, $future_id, $id, @rest) = @_;
 357   Dlog_trace { "got a receive_call for object '$id' for connection $_" } $self->_id;
 358   my $future = $self->_id_to_remote_object($future_id);
 359   $future->{method} = 'call_discard_free';
 360   my $local = $self->local_objects_by_id->{$id}
 361     or do { $future->fail("No such object $id"); return };
 362   $self->_invoke($future, $local, @rest);
 363 }
 364
 365 sub receive_call_free {
 366   my ($self, $future, $id, @rest) = @_;
 367   Dlog_trace { "got a receive_call_free for object '$id' for connection $_" } $self->_id;
 368   $self->receive_call($future, $id, undef, @rest);
 369   $self->receive_free($id);
 370 }
 371
 372 sub _invoke {
 373   my ($self, $future, $local, $ctx, $method, @args) = @_;
 374   Dlog_trace { "got _invoke for a method named '$method' for connection $_" } $self->_id;
 375   if ($method =~ /^start::/) {
 376     my $f = $local->$method(@args);
 377     $f->on_done(sub { undef($f); $future->done(@_) });
 378     return unless $f;
 379     $f->on_fail(sub { undef($f); $future->fail(@_) });
 380     return;
 381   }
 382   my $do = sub { $local->$method(@args) };
 383   eval {
 384     $future->done(
 385       defined($ctx)
 386         ? ($ctx ? $do->() : scalar($do->()))
 387         : do { $do->(); () }
 388     );
 389     1;
 390   } or do { $future->fail($@); return; };
 391   return;
 392 }
 393
 394 1;
 395
 396 =head1 NAME
 397
 398 Object::Remote::Connection - An underlying connection for L<Object::Remote>
 399
 400 =head1 LAME
 401
 402 Shipping prioritised over writing this part up. Blame mst.
 403
 404 =cut