[ViewVC] Diff of: cvs/AnyEvent-MP/MP.pm

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.86 by root, Wed Sep 9 01:47:01 2009 UTC vs.
Revision 1.103 by root, Sat Oct 17 01:42:39 2009 UTC

 =head1 NAME
-AnyEvent::MP - multi-processing/message-passing framework
+AnyEvent::MP - erlang-style multi-processing/message-passing framework
 =head1 SYNOPSIS
    use AnyEvent::MP;
    rcv $port, pong => sub { warn "pong received\n" };
    # create a port on another node
    my $port = spawn $node, $initfunc, @initdata;
+   # destroy a prot again
+   kil $port;  # "normal" kill
+   kil $port, my_error => "everything is broken"; # error kill
    # monitoring
-   mon $port, $cb->(@msg)      # callback is invoked on death
+   mon $localport, $cb->(@msg)      # callback is invoked on death
-   mon $port, $otherport       # kill otherport on abnormal death
+   mon $localport, $otherport       # kill otherport on abnormal death
-   mon $port, $otherport, @msg # send message on death
+   mon $localport, $otherport, @msg # send message on death
+   # temporarily execute code in port context
+   peval $port, sub { die "kill the port!" };
+   # execute callbacks in $SELF port context
+   my $timer = AE::timer 1, 0, psub {
+      die "kill the port, delayed";
+   };
 =head1 CURRENT STATUS
    bin/aemp                - stable.
    AnyEvent::MP            - stable API, should work.
    AnyEvent::MP::Intro     - explains most concepts.
-   AnyEvent::MP::Kernel    - mostly stable.
+   AnyEvent::MP::Kernel    - mostly stable API.
-   AnyEvent::MP::Global    - stable but incomplete, protocol not yet final.
+   AnyEvent::MP::Global    - stable API.
-stay tuned.
 =head1 DESCRIPTION
 This module (-family) implements a simple message passing framework.
 seed node that blocks for long periods will slow down everybody else.
 =item seeds - C<host:port>
 Seeds are transport endpoint(s) (usually a hostname/IP address and a
-TCP port) of nodes thta should be used as seed nodes.
+TCP port) of nodes that should be used as seed nodes.
 The nodes listening on those endpoints are expected to be long-running,
 and at least one of those should always be available. When nodes run out
 of connections (e.g. due to a network error), they try to re-establish
 connections to some seednodes again to join the network.
 use AE ();
 use base "Exporter";
-our $VERSION = $AnyEvent::MP::Kernel::VERSION;
+our $VERSION = 1.22;
 our @EXPORT = qw(
    NODE $NODE *SELF node_of after
    configure
-   snd rcv mon mon_guard kil reg psub spawn
+   snd rcv mon mon_guard kil psub peval spawn cal
    port
 );
 our $SELF;
 L<AnyEvent::MP::Global> module, which will then use it to keep
 connectivity with at least one node at any point in time.
 =back
-Example: become a distributed node using the locla node name as profile.
+Example: become a distributed node using the local node name as profile.
 This should be the most common form of invocation for "daemon"-type nodes.
    configure
 Example: become an anonymous node. This form is often used for commandline
          msg1 => sub { ... },
          ...
    ;
 Example: temporarily register a rcv callback for a tag matching some port
-(e.g. for a rpc reply) and unregister it after a message was received.
+(e.g. for an rpc reply) and unregister it after a message was received.
    rcv $port, $otherport => sub {
       my @reply = @_;
       rcv $SELF, $otherport;
       if (ref $_[0]) {
          if (my $self = $PORT_DATA{$portid}) {
             "AnyEvent::MP::Port" eq ref $self
                or Carp::croak "$port: rcv can only be called on message matching ports, caught";
-            $self->[2] = shift;
+            $self->[0] = shift;
          } else {
             my $cb = shift;
             $PORT{$portid} = sub {
                local $SELF = $port;
                eval { &$cb }; _self_die if $@;
             };
          }
       } elsif (defined $_[0]) {
          my $self = $PORT_DATA{$portid} ||= do {
-            my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port";
+            my $self = bless [$PORT{$portid} || sub { }, { }, $port], "AnyEvent::MP::Port";
             $PORT{$portid} = sub {
                local $SELF = $port;
                if (my $cb = $self->[1]{$_[0]}) {
    }
    $port
 }
+=item peval $port, $coderef[, @args]
+Evaluates the given C<$codref> within the contetx of C<$port>, that is,
+when the code throews an exception the C<$port> will be killed.
+Any remaining args will be passed to the callback. Any return values will
+be returned to the caller.
+This is useful when you temporarily want to execute code in the context of
+a port.
+Example: create a port and run some initialisation code in it's context.
+   my $port = port { ... };
+   peval $port, sub {
+      init
+         or die "unable to init";
+   };
+=cut
+sub peval($$) {
+   local $SELF = shift;
+   my $cb = shift;
+   if (wantarray) {
+      my @res = eval { &$cb };
+      _self_die if $@;
+      @res
+   } else {
+      my $res = eval { &$cb };
+      _self_die if $@;
+      $res
+   }
+}
 =item $closure = psub { BLOCK }
 Remembers C<$SELF> and creates a closure out of the BLOCK. When the
 closure is executed, sets up the environment in the same way as in C<rcv>
 callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
+The effect is basically as if it returned C<< sub { peval $SELF, sub {
+BLOCK } } >>.
 This is useful when you register callbacks from C<rcv> callbacks:
    rcv delayed_reply => sub {
       my ($delay, @reply) = @_;
 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 conenctions).
+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.
    }
    $node->monitor ($port, $cb);
    defined wantarray
-      and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
+      and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) })
 }
 =item $guard = mon_guard $port, $ref, $ref...
 Monitors the given C<$port> and keeps the passed references. When the port
          ? $action[0]()
          : snd @action;
    };
 }
+=item cal $port, @msg, $callback[, $timeout]
+A simple form of RPC - sends a message to the given C<$port> with the
+given contents (C<@msg>), but adds a reply port to the message.
+The reply port is created temporarily just for the purpose of receiving
+the reply, and will be C<kil>ed when no longer needed.
+A reply message sent to the port is passed to the C<$callback> as-is.
+If an optional time-out (in seconds) is given and it is not C<undef>,
+then the callback will be called without any arguments after the time-out
+elapsed and the port is C<kil>ed.
+If no time-out is given (or it is C<undef>), then the local port will
+monitor the remote port instead, so it eventually gets cleaned-up.
+Currently this function returns the temporary port, but this "feature"
+might go in future versions unless you can make a convincing case that
+this is indeed useful for something.
+=cut
+sub cal(@) {
+   my $timeout = ref $_[-1] ? undef : pop;
+   my $cb = pop;
+   my $port = port {
+      undef $timeout;
+      kil $SELF;
+      &$cb;
+   };
+   if (defined $timeout) {
+      $timeout = AE::timer $timeout, 0, sub {
+         undef $timeout;
+         kil $port;
+         $cb->();
+      };
+   } else {
+      mon $_[0], sub {
+         kil $port;
+         $cb->();
+      };
+   }
+   push @_, $port;
+   &snd;
+   $port
+}
 =back
 =head1 AnyEvent::MP vs. Distributed Erlang
 AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
 == aemp node, Erlang process == aemp port), so many of the documents and
 programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
 sample:
-   http://www.Erlang.se/doc/programming_rules.shtml
+   http://www.erlang.se/doc/programming_rules.shtml
-   http://Erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
+   http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
-   http://Erlang.org/download/Erlang-book-part1.pdf      # chapters 5 and 6
+   http://erlang.org/download/erlang-book-part1.pdf      # chapters 5 and 6
-   http://Erlang.org/download/armstrong_thesis_2003.pdf  # chapters 4 and 5
+   http://erlang.org/download/armstrong_thesis_2003.pdf  # chapters 4 and 5
 Despite the similarities, there are also some important differences:
 =over 4
 =item * Node IDs are arbitrary strings in AEMP.
 Erlang relies on special naming and DNS to work everywhere in the same
 way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
-configuration or DNS), but will otherwise discover other odes itself.
+configuration or DNS), and possibly the addresses of some seed nodes, but
+will otherwise discover other nodes (and their IDs) itself.
 =item * Erlang has a "remote ports are like local ports" philosophy, AEMP
 uses "local ports are like remote ports".
 The failure modes for local ports are quite different (runtime errors
 so does not need a queue that can overflow). AEMP sends are immediate,
 connection establishment is handled in the background.
 =item * Erlang suffers from silent message loss, AEMP does not.
-Erlang makes few guarantees on messages delivery - messages can get lost
+Erlang implements few guarantees on messages delivery - messages can get
-without any of the processes realising it (i.e. you send messages a, b,
+lost without any of the processes realising it (i.e. you send messages a,
-and c, and the other side only receives messages a and c).
+b, and c, and the other side only receives messages a and c).
 AEMP guarantees correct ordering, and the guarantee that after one message
 is lost, all following ones sent to the same port are lost as well, until
 monitoring raises an error, so there are no silent "holes" in the message
 sequence.

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing AnyEvent-MP/MP.pm (file contents): Revision 1.86 by root, Wed Sep 9 01:47:01 2009 UTC vs. Revision 1.103 by root, Sat Oct 17 01:42:39 2009 UTC

Diff Legend

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.86 by root, Wed Sep 9 01:47:01 2009 UTC vs.
Revision 1.103 by root, Sat Oct 17 01:42:39 2009 UTC