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     __remote_tied_hash__ => sub {
 124         my %tied_hash;
 125         tie %tied_hash, 'Object::Remote::Tied', $self->_id_to_remote_object(@_);
 126         return \%tied_hash;
 127     }
 128   )->filter_json_single_key_object(
 129     __remote_tied_array__ => sub {
 130         my @tied_array;
 131         tie @tied_array, 'Object::Remote::Tied', $self->_id_to_remote_object(@_);
 132         return \@tied_array;
 133     }
 134   );
 135 }
 136
 137 BEGIN {
 138   unshift our @Guess, sub { blessed($_[0]) ? $_[0] : undef };
 139   eval { require Object::Remote::Connector::Local };
 140   eval { require Object::Remote::Connector::LocalSudo };
 141   eval { require Object::Remote::Connector::SSH };
 142   eval { require Object::Remote::Connector::UNIX };
 143 }
 144
 145 sub new_from_spec {
 146   my ($class, $spec) = @_;
 147   return $spec if blessed $spec;
 148   Dlog_debug { "creating a new connection from spec" };
 149   foreach my $poss (do { our @Guess }) {
 150     if (my $conn = $poss->($spec)) {
 151       return $conn->maybe::start::connect;
 152     }
 153   }
 154   die "Couldn't figure out what to do with ${spec}";
 155 }
 156
 157 sub remote_object {
 158   my ($self, @args) = @_;
 159   Object::Remote::Handle->new(
 160     connection => $self, @args
 161   )->proxy;
 162 }
 163
 164 sub connect {
 165   my ($self, $to) = @_;
 166   Dlog_debug { "Creating connection to remote node '$to' for connection $_" } $self->_id;
 167   return await_future(
 168     $self->send_class_call(0, 'Object::Remote', connect => $to)
 169   );
 170 }
 171
 172 sub remote_sub {
 173   my ($self, $sub) = @_;
 174   my ($pkg, $name) = $sub =~ m/^(.*)::([^:]+)$/;
 175   Dlog_debug { "Invoking remote sub '$sub' for connection $_" } $self->_id;
 176   return await_future($self->send_class_call(0, $pkg, can => $name));
 177 }
 178
 179 sub send_class_call {
 180   my ($self, $ctx, @call) = @_;
 181   Dlog_trace { "Sending a class call for connection $_" } $self->_id;
 182   $self->send(call => class_call_handler => $ctx => call => @call);
 183 }
 184
 185 sub register_class_call_handler {
 186   my ($self) = @_;
 187   $self->local_objects_by_id->{'class_call_handler'} ||= do {
 188     my $o = $self->new_class_call_handler;
 189     $self->_local_object_to_id($o);
 190     $o;
 191   };
 192 }
 193
 194 sub new_class_call_handler {
 195   Object::Remote::CodeContainer->new(
 196     code => sub {
 197       my ($class, $method) = (shift, shift);
 198       use_module($class)->$method(@_);
 199     }
 200   );
 201 }
 202
 203 sub register_remote {
 204   my ($self, $remote) = @_;
 205   Dlog_trace { my $i = $remote->id; "Registered a remote object with id of '$i' for connection $_" } $self->_id;
 206   weaken($self->remote_objects_by_id->{$remote->id} = $remote);
 207   return $remote;
 208 }
 209
 210 sub send_free {
 211   my ($self, $id) = @_;
 212   Dlog_trace { "sending request to free object '$id' for connection $_" } $self->_id;
 213   delete $self->remote_objects_by_id->{$id};
 214   $self->_send([ free => $id ]);
 215 }
 216
 217 sub send {
 218   my ($self, $type, @call) = @_;
 219
 220   my $future = CPS::Future->new;
 221   my $remote = $self->remote_objects_by_id->{$call[0]};
 222
 223   unshift @call, $type => $self->_local_object_to_id($future);
 224
 225   my $outstanding = $self->outstanding_futures;
 226   $outstanding->{$future} = $future;
 227   $future->on_ready(sub {
 228     undef($remote);
 229     delete $outstanding->{$future}
 230   });
 231
 232   $self->_send(\@call);
 233
 234   return $future;
 235 }
 236
 237 sub send_discard {
 238   my ($self, $type, @call) = @_;
 239
 240   unshift @call, $type => 'NULL';
 241
 242   $self->_send(\@call);
 243 }
 244
 245 sub _send {
 246   my ($self, $to_send) = @_;
 247   my $fh = $self->send_to_fh;
 248   Dlog_trace { "Starting to serialize data in argument to _send for connection $_" } $self->_id;
 249   my $serialized = $self->_serialize($to_send)."\n";
 250   Dlog_trace { my $l = length($serialized); "serialization is completed; sending '$l' characters of serialized data to $_" } $fh;
 251   #TODO this is very risky for deadlocks unless it's set to non-blocking and then with out extra
 252   #logic it could easily do short-writes to the remote side - how about taking this entire buffer
 253   #and having the run loop send it to the file handle so this doesn't block while the sending
 254   #is happening?
 255   my $ret = print $fh $serialized;
 256   Dlog_trace { my $r = defined $ret ? $ret : 'undef'; "print() returned $r with $_" } $fh;
 257   #TODO hrm reason print's return value was ignored?
 258   die "could not write to filehandle: $!" unless $ret;
 259   return $ret;
 260 }
 261
 262 sub _serialize {
 263   my ($self, $data) = @_;
 264   local our @New_Ids = (-1);
 265   return eval {
 266     my $flat = $self->_encode($self->_deobjectify($data));
 267     warn "$$ >>> ${flat}\n" if $DEBUG;
 268     $flat;
 269   } || do {
 270     my $err = $@; # won't get here if the eval doesn't die
 271     # don't keep refs to new things
 272     delete @{$self->local_objects_by_id}{@New_Ids};
 273     die "Error serializing: $err";
 274   };
 275 }
 276
 277 sub _local_object_to_id {
 278   my ($self, $object) = @_;
 279   my $id = refaddr($object);
 280   $self->local_objects_by_id->{$id} ||= do {
 281     push our(@New_Ids), $id if @New_Ids;
 282     $object;
 283   };
 284   return $id;
 285 }
 286
 287 sub _deobjectify {
 288   my ($self, $data) = @_;
 289   if (blessed($data)) {
 290     return +{ __remote_object__ => $self->_local_object_to_id($data) };
 291   } elsif (my $ref = ref($data)) {
 292     if ($ref eq 'HASH') {
 293       my $tied_to = tied(%$data);
 294       if(defined($tied_to)) {
 295         return +{__remote_tied_hash__ => $self->_local_object_to_id($tied_to)};
 296       } else {
 297         return +{ map +($_ => $self->_deobjectify($data->{$_})), keys %$data };
 298       }
 299     } elsif ($ref eq 'ARRAY') {
 300       my $tied_to = tied(@$data);
 301       if (defined($tied_to)) {
 302         return +{__remote_tied_array__ => $self->_local_object_to_id($tied_to)};
 303       } else {
 304         return [ map $self->_deobjectify($_), @$data ];
 305       }
 306     } elsif ($ref eq 'CODE') {
 307       my $id = $self->_local_object_to_id(
 308                  Object::Remote::CodeContainer->new(code => $data)
 309                );
 310       return +{ __remote_code__ => $id };
 311     } elsif ($ref eq 'SCALAR') {
 312       return +{ __scalar_ref__ => $$data };
 313     } elsif ($ref eq 'GLOB') {
 314       return +{ __glob_ref__ => $self->_local_object_to_id(
 315         Object::Remote::GlobContainer->new(handle => $data)
 316       ) };
 317     } else {
 318       die "Can't collapse reftype $ref";
 319     }
 320   }
 321   return $data; # plain scalar
 322 }
 323
 324 sub _receive {
 325   my ($self, $flat) = @_;
 326   warn "$$ <<< $flat\n" if $DEBUG;
 327   Dlog_trace { my $l = length($flat); "Starting to deserialize $l characters of data for connection $_" } $self->_id;
 328   my ($type, @rest) = eval { @{$self->_deserialize($flat)} }
 329     or do { warn "Deserialize failed for ${flat}: $@"; return };
 330   Dlog_trace { "deserialization complete for connection $_" } $self->_id;
 331   eval { $self->${\"receive_${type}"}(@rest); 1 }
 332     or do { warn "Receive failed for ${flat}: $@"; return };
 333   return;
 334 }
 335
 336 sub receive_free {
 337   my ($self, $id) = @_;
 338   Dlog_trace { "got a receive_free for object '$id' for connection $_" } $self->_id;
 339   delete $self->local_objects_by_id->{$id}
 340     or warn "Free: no such object $id";
 341   return;
 342 }
 343
 344 sub receive_call {
 345   my ($self, $future_id, $id, @rest) = @_;
 346   Dlog_trace { "got a receive_call for object '$id' for connection $_" } $self->_id;
 347   my $future = $self->_id_to_remote_object($future_id);
 348   $future->{method} = 'call_discard_free';
 349   my $local = $self->local_objects_by_id->{$id}
 350     or do { $future->fail("No such object $id"); return };
 351   $self->_invoke($future, $local, @rest);
 352 }
 353
 354 sub receive_call_free {
 355   my ($self, $future, $id, @rest) = @_;
 356   Dlog_trace { "got a receive_call_free for object '$id' for connection $_" } $self->_id;
 357   $self->receive_call($future, $id, undef, @rest);
 358   $self->receive_free($id);
 359 }
 360
 361 sub _invoke {
 362   my ($self, $future, $local, $ctx, $method, @args) = @_;
 363   Dlog_trace { "got _invoke for a method named '$method' for connection $_" } $self->_id;
 364   if ($method =~ /^start::/) {
 365     my $f = $local->$method(@args);
 366     $f->on_done(sub { undef($f); $future->done(@_) });
 367     return unless $f;
 368     $f->on_fail(sub { undef($f); $future->fail(@_) });
 369     return;
 370   }
 371   my $do = sub { $local->$method(@args) };
 372   eval {
 373     $future->done(
 374       defined($ctx)
 375         ? ($ctx ? $do->() : scalar($do->()))
 376         : do { $do->(); () }
 377     );
 378     1;
 379   } or do { $future->fail($@); return; };
 380   return;
 381 }
 382
 383 1;
 384
 385 =head1 NAME
 386
 387 Object::Remote::Connection - An underlying connection for L<Object::Remote>
 388
 389 =head1 LAME
 390
 391 Shipping prioritised over writing this part up. Blame mst.
 392
 393 =cut