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

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.30 by root, Tue Aug 4 23:35:51 2009 UTC vs.
Revision 1.35 by root, Thu Aug 6 10:21:48 2009 UTC

    # more, smarter, matches (_any_ is exported by this module)
    rcv $port, [child_died => $pid] => sub { ...
    rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3
+   # linking two ports, so they both crash together
+   lnk $port1, $port2;
+   # monitoring
+   mon $port, $cb->(@msg)      # callback is invoked on death
+   mon $port, $otherport       # kill otherport on abnormal death
+   mon $port, $otherport, @msg # send message on death
 =head1 DESCRIPTION
 This module (-family) implements a simple message passing framework.
 Despite its simplicity, you can securely message other processes running
 use base "Exporter";
 our $VERSION = '0.1';
 our @EXPORT = qw(
    NODE $NODE *SELF node_of _any_
-   resolve_node
+   resolve_node initialise_node
-   become_slave become_public
    snd rcv mon kil reg psub
    port
 );
 our $SELF;
 The C<NODE> function returns, and the C<$NODE> variable contains
 the noderef of the local node. The value is initialised by a call
 to C<become_public> or C<become_slave>, after which all local port
 identifiers become invalid.
-=item $noderef = node_of $portid
+=item $noderef = node_of $port
 Extracts and returns the noderef from a portid or a noderef.
+=item initialise_node $noderef, $seednode, $seednode...
+=item initialise_node "slave/", $master, $master...
+Before a node can talk to other nodes on the network it has to initialise
+itself - the minimum a node needs to know is it's own name, and optionally
+it should know the noderefs of some other nodes in the network.
+This function initialises a node - it must be called exactly once (or
+never) before calling other AnyEvent::MP functions.
+All arguments are noderefs, which can be either resolved or unresolved.
+There are two types of networked nodes, public nodes and slave nodes:
+=over 4
+=item public nodes
+For public nodes, C<$noderef> must either be a (possibly unresolved)
+noderef, in which case it will be resolved, or C<undef> (or missing), in
+which case the noderef will be guessed.
+Afterwards, the node will bind itself on all endpoints and try to connect
+to all additional C<$seednodes> that are specified. Seednodes are optional
+and can be used to quickly bootstrap the node into an existing network.
+=item slave nodes
+When the C<$noderef> is the special string C<slave/>, then the node will
+become a slave node. Slave nodes cannot be contacted from outside and will
+route most of their traffic to the master node that they attach to.
+At least one additional noderef is required: The node will try to connect
+to all of them and will become a slave attached to the first node it can
+successfully connect to.
+=back
+This function will block until all nodes have been resolved and, for slave
+nodes, until it has successfully established a connection to a master
+server.
+Example: become a public node listening on the default node.
+   initialise_node;
+Example: become a public node, and try to contact some well-known master
+servers to become part of the network.
+   initialise_node undef, "master1", "master2";
+Example: become a public node listening on port C<4041>.
+   initialise_node 4041;
+Example: become a public node, only visible on localhost port 4044.
+   initialise_node "locahost:4044";
+Example: become a slave node to any of the specified master servers.
+   initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
 =item $cv = resolve_node $noderef
 Takes an unresolved node reference that may contain hostnames and
 abbreviated IDs, resolves all of them and returns a resolved node
 Due to some quirks in how perl exports variables, it is impossible to
 just export C<$SELF>, all the symbols called C<SELF> are exported by this
 module, but only C<$SELF> is currently used.
-=item snd $portid, type => @data
+=item snd $port, type => @data
-=item snd $portid, @msg
+=item snd $port, @msg
 Send the given message to the given port ID, which can identify either
 a local or a remote port, and can be either a string or soemthignt hat
 stringifies a sa port ID (such as a port object :).
 JSON is used, then only strings, numbers and arrays and hashes consisting
 of those are allowed (no objects). When Storable is used, then anything
 that Storable can serialise and deserialise is allowed, and for the local
 node, anything can be passed.
-=item kil $portid[, @reason]
+=item $local_port = port
-Kill the specified port with the given C<@reason>.
+Create a new local port object that can be used either as a pattern
+matching port ("full port") or a single-callback port ("miniport"),
+depending on how C<rcv> callbacks are bound to the object.
-If no C<@reason> is specified, then the port is killed "normally" (linked
+=item $port = port { my @msg = @_; $finished }
-ports will not be kileld, or even notified).
-Otherwise, linked ports get killed with the same reason (second form of
+Creates a "miniport", that is, a very lightweight port without any pattern
-C<mon>, see below).
+matching behind it, and returns its ID. Semantically the same as creating
+a port and calling C<rcv $port, $callback> on it.
-Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
+The block will be called for every message received on the port. When the
-will be reported as reason C<< die => $@ >>.
+callback returns a true value its job is considered "done" and the port
+will be destroyed. Otherwise it will stay alive.
-Transport/communication errors are reported as C<< transport_error =>
+The message will be passed as-is, no extra argument (i.e. no port id) will
-$message >>.
+be passed to the callback.
+If you need the local port id in the callback, this works nicely:
+   my $port; $port = port {
+      snd $otherport, reply => $port;
+   };
+=cut
+sub rcv($@);
+sub port(;&) {
+   my $id = "$UNIQ." . $ID++;
+   my $port = "$NODE#$id";
+   if (@_) {
+      rcv $port, shift;
+   } else {
+      $PORT{$id} = sub { }; # nop
+   }
+   $port
+}
+=item reg $port, $name
+Registers the given port under the name C<$name>. If the name already
+exists it is replaced.
+A port can only be registered under one well known name.
+A port automatically becomes unregistered when it is killed.
+=cut
+sub reg(@) {
+   my ($port, $name) = @_;
+   $REG{$name} = $port;
+}
+=item rcv $port, $callback->(@msg)
+Replaces the callback on the specified miniport (after converting it to
+one if required).
+=item rcv $port, tagstring        => $callback->(@msg), ...
+=item rcv $port, $smartmatch      => $callback->(@msg), ...
+=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
+Register callbacks to be called on matching messages on the given full
+port (after converting it to one if required).
+The callback has to return a true value when its work is done, after
+which is will be removed, or a false value in which case it will stay
+registered.
+The global C<$SELF> (exported by this module) contains C<$port> while
+executing the callback.
+Runtime errors wdurign callback execution will result in the port being
+C<kil>ed.
+If the match is an array reference, then it will be matched against the
+first elements of the message, otherwise only the first element is being
+matched.
+Any element in the match that is specified as C<_any_> (a function
+exported by this module) matches any single element of the message.
+While not required, it is highly recommended that the first matching
+element is a string identifying the message. The one-string-only match is
+also the most efficient match (by far).
+=cut
+sub rcv($@) {
+   my $port = shift;
+   my ($noderef, $portid) = split /#/, $port, 2;
+   ($NODE{$noderef} || add_node $noderef) == $NODE{""}
+      or Carp::croak "$port: rcv can only be called on local ports, caught";
+   if (@_ == 1) {
+      my $cb = shift;
+      delete $PORT_DATA{$portid};
+      $PORT{$portid} = sub {
+         local $SELF = $port;
+         eval {
+            &$cb
+               and kil $port;
+         };
+         _self_die if $@;
+      };
+   } else {
+      my $self = $PORT_DATA{$portid} ||= do {
+         my $self = bless {
+            id    => $port,
+         }, "AnyEvent::MP::Port";
+         $PORT{$portid} = sub {
+            local $SELF = $port;
+            eval {
+               for (@{ $self->{rc0}{$_[0]} }) {
+                  $_ && &{$_->[0]}
+                     && undef $_;
+               }
+               for (@{ $self->{rcv}{$_[0]} }) {
+                  $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
+                     && &{$_->[0]}
+                     && undef $_;
+               }
+               for (@{ $self->{any} }) {
+                  $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
+                     && &{$_->[0]}
+                     && undef $_;
+               }
+            };
+            _self_die if $@;
+         };
+         $self
+      };
+      "AnyEvent::MP::Port" eq ref $self
+         or Carp::croak "$port: rcv can only be called on message matching ports, caught";
+      while (@_) {
+         my ($match, $cb) = splice @_, 0, 2;
+         if (!ref $match) {
+            push @{ $self->{rc0}{$match}      }, [$cb];
+         } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
+            my ($type, @match) = @$match;
+            @match
+               ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
+               : push @{ $self->{rc0}{$match->[0]} }, [$cb];
+         } else {
+            push @{ $self->{any}              }, [$cb, $match];
+         }
+      }
+   }
+   $port
+}
+=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.
+This is useful when you register callbacks from C<rcv> callbacks:
+   rcv delayed_reply => sub {
+      my ($delay, @reply) = @_;
+      my $timer = AE::timer $delay, 0, psub {
+         snd @reply, $SELF;
+      };
+   };
+=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 $portid, $cb->(@reason)
+=item $guard = mon $port, $cb->(@reason)
-=item $guard = mon $portid, $otherport
+=item $guard = mon $port, $otherport
-=item $guard = mon $portid, $otherport, @msg
+=item $guard = mon $port, $otherport, @msg
 Monitor the given port and do something when the port is killed.
 In the first form, the callback is simply called with any number
 of C<@reason> elements (no @reason means that the port was deleted
    mon $port2, $port1;
 It means that if either one is killed abnormally, the other one gets
 killed as well.
-=item $local_port = port
+=item kil $port[, @reason]
-Create a new local port object that supports message matching.
+Kill the specified port with the given C<@reason>.
-=item $portid = port { my @msg = @_; $finished }
+If no C<@reason> is specified, then the port is killed "normally" (linked
+ports will not be kileld, or even notified).
-Creates a "mini port", that is, a very lightweight port without any
+Otherwise, linked ports get killed with the same reason (second form of
-pattern matching behind it, and returns its ID.
+C<mon>, see below).
-The block will be called for every message received on the port. When the
+Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
-callback returns a true value its job is considered "done" and the port
+will be reported as reason C<< die => $@ >>.
-will be destroyed. Otherwise it will stay alive.
-The message will be passed as-is, no extra argument (i.e. no port id) will
+Transport/communication errors are reported as C<< transport_error =>
-be passed to the callback.
+$message >>.
-If you need the local port id in the callback, this works nicely:
-   my $port; $port = miniport {
-      snd $otherport, reply => $port;
-   };
-=cut
-sub port(;&) {
-   my $id = "$UNIQ." . $ID++;
-   my $port = "$NODE#$id";
-   if (@_) {
-      my $cb = shift;
-      $PORT{$id} = sub {
-         local $SELF = $port;
-         eval {
-            &$cb
-               and kil $id;
-         };
-         _self_die if $@;
-      };
-   } else {
-      my $self = bless {
-         id    => "$NODE#$id",
-      }, "AnyEvent::MP::Port";
-      $PORT_DATA{$id} = $self;
-      $PORT{$id} = sub {
-         local $SELF = $port;
-         eval {
-            for (@{ $self->{rc0}{$_[0]} }) {
-               $_ && &{$_->[0]}
-                  && undef $_;
-            }
-            for (@{ $self->{rcv}{$_[0]} }) {
-               $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
-                  && &{$_->[0]}
-                  && undef $_;
-            }
-            for (@{ $self->{any} }) {
-               $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
-                  && &{$_->[0]}
-                  && undef $_;
-            }
-         };
-         _self_die if $@;
-      };
-   }
-   $port
-}
-=item reg $portid, $name
-Registers the given port under the name C<$name>. If the name already
-exists it is replaced.
-A port can only be registered under one well known name.
-A port automatically becomes unregistered when it is killed.
-=cut
-sub reg(@) {
-   my ($portid, $name) = @_;
-   $REG{$name} = $portid;
-}
-=item rcv $portid, tagstring        => $callback->(@msg), ...
-=item rcv $portid, $smartmatch      => $callback->(@msg), ...
-=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
-Register callbacks to be called on matching messages on the given port.
-The callback has to return a true value when its work is done, after
-which is will be removed, or a false value in which case it will stay
-registered.
-The global C<$SELF> (exported by this module) contains C<$portid> while
-executing the callback.
-Runtime errors wdurign callback execution will result in the port being
-C<kil>ed.
-If the match is an array reference, then it will be matched against the
-first elements of the message, otherwise only the first element is being
-matched.
-Any element in the match that is specified as C<_any_> (a function
-exported by this module) matches any single element of the message.
-While not required, it is highly recommended that the first matching
-element is a string identifying the message. The one-string-only match is
-also the most efficient match (by far).
-=cut
-sub rcv($@) {
-   my ($noderef, $port) = split /#/, shift, 2;
-   ($NODE{$noderef} || add_node $noderef) == $NODE{""}
-      or Carp::croak "$noderef#$port: rcv can only be called on local ports, caught";
-   my $self = $PORT_DATA{$port}
-      or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
-   "AnyEvent::MP::Port" eq ref $self
-      or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
-   while (@_) {
-      my ($match, $cb) = splice @_, 0, 2;
-      if (!ref $match) {
-         push @{ $self->{rc0}{$match}      }, [$cb];
-      } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
-         my ($type, @match) = @$match;
-         @match
-            ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
-            : push @{ $self->{rc0}{$match->[0]} }, [$cb];
-      } else {
-         push @{ $self->{any}              }, [$cb, $match];
-      }
-   }
-}
-=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.
-This is useful when you register callbacks from C<rcv> callbacks:
-   rcv delayed_reply => sub {
-      my ($delay, @reply) = @_;
-      my $timer = AE::timer $delay, 0, psub {
-         snd @reply, $SELF;
-      };
-   };
-=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
-      }
-   }
-}
-=back
-=head1 FUNCTIONS FOR NODES
-=over 4
-=item become_public $noderef
-Tells the node to become a public node, i.e. reachable from other nodes.
-The first argument is the (unresolved) node reference of the local node
-(if missing then the empty string is used).
-It is quite common to not specify anything, in which case the local node
-tries to listen on the default port, or to only specify a port number, in
-which case AnyEvent::MP tries to guess the local addresses.
-=cut
 =back
 =head1 NODE MESSAGES
 =back
 =head1 AnyEvent::MP vs. Distributed Erlang
-AnyEvent::MP got lots of its ideas from distributed erlang (erlang node
+AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
-== aemp node, erlang process == aemp port), so many of the documents and
+== aemp node, Erlang process == aemp port), so many of the documents and
-programming techniques employed by erlang apply to AnyEvent::MP. Here is a
+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
 Erlang uses processes that selctively receive messages, and therefore
 needs a queue. AEMP is event based, queuing messages would serve no useful
 purpose.
-(But see L<Coro::MP> for a more erlang-like process model on top of AEMP).
+(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
 =item * Erlang sends are synchronous, AEMP sends are asynchronous.
-Sending messages in erlang is synchronous and blocks the process. AEMP
+Sending messages in Erlang is synchronous and blocks the process. AEMP
 sends are immediate, connection establishment is handled in the
 background.
 =item * Erlang can silently lose messages, AEMP cannot.
 and c, and the other side only receives messages a and c).
 AEMP guarantees correct ordering, and the guarantee that there are no
 holes in the message sequence.
-=item * In erlang, processes can be declared dead and later be found to be
+=item * In Erlang, processes can be declared dead and later be found to be
 alive.
-In erlang it can happen that a monitored process is declared dead and
+In Erlang it can happen that a monitored process is declared dead and
 linked processes get killed, but later it turns out that the process is
 still alive - and can receive messages.
 In AEMP, when port monitoring detects a port as dead, then that port will
 eventually be killed - it cannot happen that a node detects a port as dead
 and then later sends messages to it, finding it is still alive.
 =item * Erlang can send messages to the wrong port, AEMP does not.
-In erlang it is quite possible that a node that restarts reuses a process
+In Erlang it is quite possible that a node that restarts reuses a process
 ID known to other nodes for a completely different process, causing
 messages destined for that process to end up in an unrelated process.
 AEMP never reuses port IDs, so old messages or old port IDs floating
 around in the network will not be sent to an unrelated port.
 securely authenticate nodes.
 =item * The AEMP protocol is optimised for both text-based and binary
 communications.
-The AEMP protocol, unlike the erlang protocol, supports both
+The AEMP protocol, unlike the Erlang protocol, supports both
 language-independent text-only protocols (good for debugging) and binary,
 language-specific serialisers (e.g. Storable).
 It has also been carefully designed to be implementable in other languages
 with a minimum of work while gracefully degrading fucntionality to make the
 protocol simple.
+=item * AEMP has more flexible monitoring options than Erlang.
+In Erlang, you can chose to receive I<all> exit signals as messages
+or I<none>, there is no in-between, so monitoring single processes is
+difficult to implement. Monitoring in AEMP is more flexible than in
+Erlang, as one can choose between automatic kill, exit message or callback
+on a per-process basis.
+=item * Erlang has different semantics for monitoring and linking, AEMP has the same.
+Monitoring in Erlang is not an indicator of process death/crashes,
+as linking is (except linking is unreliable in Erlang). In AEMP, the
+semantics of monitoring and linking are identical, linking is simply
+two-way monitoring with automatic kill.
 =back
 =head1 SEE ALSO
 L<AnyEvent>.

Diff Legend

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

Comparing AnyEvent-MP/MP.pm (file contents): Revision 1.30 by root, Tue Aug 4 23:35:51 2009 UTC vs. Revision 1.35 by root, Thu Aug 6 10:21:48 2009 UTC

Diff Legend

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.30 by root, Tue Aug 4 23:35:51 2009 UTC vs.
Revision 1.35 by root, Thu Aug 6 10:21:48 2009 UTC