view release on metacpan or  search on metacpan

MP.pm  view on Meta::CPAN

=cut

sub psub(&) {
   my $cb = shift;

   my $port = $SELF
      or Carp::croak "psub can only be called from within rcv or psub callbacks, not";

   sub {
      local $SELF = $port;

      if (wantarray) {
         my @res = eval { &$cb };
         _self_die if $@;
         @res
      } else {
         my $res = eval { &$cb };
         _self_die if $@;
         $res
      }
   }
}

=item $guard = mon $port, $rcvport          # kill $rcvport when $port dies

=item $guard = mon $port                    # kill $SELF when $port dies

=item $guard = mon $port, $cb->(@reason)    # call $cb when $port dies

=item $guard = mon $port, $rcvport, @msg    # send a message when $port dies

Monitor the given port and do something when the port is killed or
messages to it were lost, and optionally return a guard that can be used
to stop monitoring again.

The first two forms distinguish between "normal" and "abnormal" kil's:

In the first form (another port given), if the C<$port> is C<kil>'ed with
a non-empty reason, the other port (C<$rcvport>) will be kil'ed with the
same reason. That is, on "normal" kil's nothing happens, while under all
other conditions, the other port is killed with the same reason.

The second form (kill self) is the same as the first form, except that
C<$rvport> defaults to C<$SELF>.

The remaining forms don't distinguish between "normal" and "abnormal" kil's
- it's up to the callback or receiver to check whether the C<@reason> is
empty and act accordingly.

In the third form (callback), the callback is simply called with any
number of C<@reason> elements (empty @reason means that the port was deleted
"normally"). Note also that I<< the callback B<must> never die >>, so use
C<eval> if unsure.

In the last form (message), a message of the form C<$rcvport, @msg,
@reason> will be C<snd>.

Monitoring-actions are one-shot: once messages are lost (and a monitoring
alert was raised), they are removed and will not trigger again, even if it
turns out that the port is still alive.

As a rule of thumb, monitoring requests should always monitor a remote
port locally (using a local C<$rcvport> or a callback). The reason is that
kill messages might get lost, just like any other message. Another less
obvious reason is that even monitoring requests can get lost (for example,
when the connection to the other node goes down permanently). When
monitoring a port locally these problems do not exist.

C<mon> effectively guarantees that, in the absence of hardware failures,
after starting the monitor, either all messages sent to the port will
arrive, or the monitoring action will be invoked after possible message
loss has been detected. No messages will be lost "in between" (after
the first lost message no further messages will be received by the
port). After the monitoring action was invoked, further messages might get
delivered again.

Inter-host-connection timeouts and monitoring depend on the transport
used. The only transport currently implemented is TCP, and AnyEvent::MP
relies on TCP to detect node-downs (this can take 10-15 minutes on a
non-idle connection, and usually around two hours for idle connections).

This means that monitoring is good for program errors and cleaning up
stuff eventually, but they are no replacement for a timeout when you need
to ensure some maximum latency.

Example: call a given callback when C<$port> is killed.

   mon $port, sub { warn "port died because of <@_>\n" };

Example: kill ourselves when C<$port> is killed abnormally.

   mon $port;

Example: send us a restart message when another C<$port> is killed.

   mon $port, $self => "restart";

=cut

sub mon {
   my ($nodeid, $port) = split /#/, shift, 2;

   my $node = $NODE{$nodeid} || add_node $nodeid;

   my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';

   unless (ref $cb) {
      if (@_) {
         # send a kill info message
         my (@msg) = ($cb, @_);
         $cb = sub { snd @msg, @_ };
      } else {
         # simply kill other port
         my $port = $cb;
         $cb = sub { kil $port, @_ if @_ };
      }
   }

   $node->monitor ($port, $cb);