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

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.85 by root, Tue Sep 8 01:54:13 2009 UTC vs.
Revision 1.114 by root, Thu Apr 22 16:06:19 2010 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.
 to have fixed listening addresses, seed nodes are perfectly normal nodes -
 any node can function as a seed node for others.
 In addition to discovering the network, seed nodes are also used to
 maintain the network and to connect nodes that otherwise would have
-trouble connecting. They form the backbone of the AnyEvent::MP network.
+trouble connecting. They form the backbone of an AnyEvent::MP network.
 Seed nodes are expected to be long-running, and at least one seed node
 should always be available. They should also be relatively responsive - a
 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.28;
 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;
 Before a node can talk to other nodes on the network (i.e. enter
 "distributed mode") it has to configure itself - the minimum a node needs
 to know is its own name, and optionally it should know the addresses of
 some other nodes in the network to discover other nodes.
+The key/value pairs are basically the same ones as documented for the
+F<aemp> command line utility (sans the set/del prefix).
 This function configures a node - it must be called exactly once (or
 never) before calling other AnyEvent::MP functions.
 =over 4
 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
 =item kil $port[, @reason]
 Kill the specified port with the given C<@reason>.
-If no C<@reason> is specified, then the port is killed "normally" (ports
+If no C<@reason> is specified, then the port is killed "normally" -
-monitoring other ports will not necessarily die because a port dies
+monitor callback will be invoked, but the kil will not cause linked ports
-"normally").
+(C<mon $mport, $lport> form) to get killed.
-Otherwise, linked ports get killed with the same reason (second form of
+If a C<@reason> is specified, then linked ports (C<mon $mport, $lport>
-C<mon>, see above).
+form) get killed with the same reason.
 Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
 will be reported as reason C<< die => $@ >>.
 Transport/communication errors are reported as C<< transport_error =>
          ? $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.
 overhead, as well as having to keep a proxy object everywhere.
 Strings can easily be printed, easily serialised etc. and need no special
 procedures to be "valid".
-And as a result, a miniport consists of a single closure stored in a
+And as a result, a port with just a default receiver consists of a single
-global hash - it can't become much cheaper.
+closure stored in a global hash - it can't become much cheaper.
 =item Why favour JSON, why not a real serialising format such as Storable?
 In fact, any AnyEvent::MP node will happily accept Storable as framing
 format, but currently there is no way to make a node use Storable by
 L<AnyEvent::MP::Intro> - a gentle introduction.
 L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
-L<AnyEvent::MP::Global> - network maintainance and port groups, to find
+L<AnyEvent::MP::Global> - network maintenance and port groups, to find
 your applications.
+L<AnyEvent::MP::DataConn> - establish data connections between nodes.
 L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
 all nodes.
 L<AnyEvent>.

Diff Legend

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

Comparing AnyEvent-MP/MP.pm (file contents): Revision 1.85 by root, Tue Sep 8 01:54:13 2009 UTC vs. Revision 1.114 by root, Thu Apr 22 16:06:19 2010 UTC

Diff Legend

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.85 by root, Tue Sep 8 01:54:13 2009 UTC vs.
Revision 1.114 by root, Thu Apr 22 16:06:19 2010 UTC