ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-MP/MP.pm
(Generate patch)

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.71 by root, Sun Aug 30 19:52:56 2009 UTC vs.
Revision 1.79 by root, Fri Sep 4 21:52:09 2009 UTC

4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::MP; 7 use AnyEvent::MP;
8 8
9 $NODE # contains this node's noderef 9 $NODE # contains this node's node ID
10 NODE # returns this node's noderef 10 NODE # returns this node's node ID
11 NODE $port # returns the noderef of the port
12 11
13 $SELF # receiving/own port id in rcv callbacks 12 $SELF # receiving/own port id in rcv callbacks
14 13
15 # initialise the node so it can send/receive messages 14 # initialise the node so it can send/receive messages
16 initialise_node; 15 configure;
17 16
18 # ports are message endpoints 17 # ports are message destinations
19 18
20 # sending messages 19 # sending messages
21 snd $port, type => data...; 20 snd $port, type => data...;
22 snd $port, @msg; 21 snd $port, @msg;
23 snd @msg_with_first_element_being_a_port; 22 snd @msg_with_first_element_being_a_port;
24 23
25 # creating/using ports, the simple way 24 # creating/using ports, the simple way
26 my $simple_port = port { my @msg = @_; 0 }; 25 my $simple_port = port { my @msg = @_ };
27 26
28 # creating/using ports, tagged message matching 27 # creating/using ports, tagged message matching
29 my $port = port; 28 my $port = port;
30 rcv $port, ping => sub { snd $_[0], "pong"; 0 }; 29 rcv $port, ping => sub { snd $_[0], "pong" };
31 rcv $port, pong => sub { warn "pong received\n"; 0 }; 30 rcv $port, pong => sub { warn "pong received\n" };
32 31
33 # create a port on another node 32 # create a port on another node
34 my $port = spawn $node, $initfunc, @initdata; 33 my $port = spawn $node, $initfunc, @initdata;
35 34
36 # monitoring 35 # monitoring
40 39
41=head1 CURRENT STATUS 40=head1 CURRENT STATUS
42 41
43 bin/aemp - stable. 42 bin/aemp - stable.
44 AnyEvent::MP - stable API, should work. 43 AnyEvent::MP - stable API, should work.
45 AnyEvent::MP::Intro - uptodate, but incomplete. 44 AnyEvent::MP::Intro - explains most concepts.
46 AnyEvent::MP::Kernel - mostly stable. 45 AnyEvent::MP::Kernel - mostly stable.
47 AnyEvent::MP::Global - stable API, protocol not yet final. 46 AnyEvent::MP::Global - stable but incomplete, protocol not yet final.
48 47
49 stay tuned. 48stay tuned.
50 49
51=head1 DESCRIPTION 50=head1 DESCRIPTION
52 51
53This module (-family) implements a simple message passing framework. 52This module (-family) implements a simple message passing framework.
54 53
56on the same or other hosts, and you can supervise entities remotely. 55on the same or other hosts, and you can supervise entities remotely.
57 56
58For an introduction to this module family, see the L<AnyEvent::MP::Intro> 57For an introduction to this module family, see the L<AnyEvent::MP::Intro>
59manual page and the examples under F<eg/>. 58manual page and the examples under F<eg/>.
60 59
61At the moment, this module family is a bit underdocumented.
62
63=head1 CONCEPTS 60=head1 CONCEPTS
64 61
65=over 4 62=over 4
66 63
67=item port 64=item port
68 65
69A port is something you can send messages to (with the C<snd> function). 66Not to be confused with a TCP port, a "port" is something you can send
67messages to (with the C<snd> function).
70 68
71Ports allow you to register C<rcv> handlers that can match all or just 69Ports allow you to register C<rcv> handlers that can match all or just
72some messages. Messages send to ports will not be queued, regardless of 70some messages. Messages send to ports will not be queued, regardless of
73anything was listening for them or not. 71anything was listening for them or not.
74 72
138 136
139our $VERSION = $AnyEvent::MP::Kernel::VERSION; 137our $VERSION = $AnyEvent::MP::Kernel::VERSION;
140 138
141our @EXPORT = qw( 139our @EXPORT = qw(
142 NODE $NODE *SELF node_of after 140 NODE $NODE *SELF node_of after
143 initialise_node 141 configure
144 snd rcv mon mon_guard kil reg psub spawn 142 snd rcv mon mon_guard kil reg psub spawn
145 port 143 port
146); 144);
147 145
148our $SELF; 146our $SELF;
155 153
156=item $thisnode = NODE / $NODE 154=item $thisnode = NODE / $NODE
157 155
158The C<NODE> function returns, and the C<$NODE> variable contains, the node 156The C<NODE> function returns, and the C<$NODE> variable contains, the node
159ID of the node running in the current process. This value is initialised by 157ID of the node running in the current process. This value is initialised by
160a call to C<initialise_node>. 158a call to C<configure>.
161 159
162=item $nodeid = node_of $port 160=item $nodeid = node_of $port
163 161
164Extracts and returns the node ID from a port ID or a node ID. 162Extracts and returns the node ID from a port ID or a node ID.
165 163
166=item initialise_node $profile_name, key => value... 164=item configure $profile, key => value...
165
166=item configure key => value...
167 167
168Before a node can talk to other nodes on the network (i.e. enter 168Before a node can talk to other nodes on the network (i.e. enter
169"distributed mode") it has to initialise itself - the minimum a node needs 169"distributed mode") it has to configure itself - the minimum a node needs
170to know is its own name, and optionally it should know the addresses of 170to know is its own name, and optionally it should know the addresses of
171some other nodes in the network to discover other nodes. 171some other nodes in the network to discover other nodes.
172 172
173This function initialises a node - it must be called exactly once (or 173This function configures a node - it must be called exactly once (or
174never) before calling other AnyEvent::MP functions. 174never) before calling other AnyEvent::MP functions.
175 175
176The first argument is a profile name. If it is C<undef> or missing, then 176=over 4
177the current nodename will be used instead (i.e. F<uname -n>).
178 177
178=item step 1, gathering configuration from profiles
179
179The function first looks up the profile in the aemp configuration (see the 180The function first looks up a profile in the aemp configuration (see the
180L<aemp> commandline utility). the profile is calculated as follows: 181L<aemp> commandline utility). The profile name can be specified via the
182named C<profile> parameter or can simply be the first parameter). If it is
183missing, then the nodename (F<uname -n>) will be used as profile name.
181 184
185The profile data is then gathered as follows:
186
182First, all remaining key => value pairs (all of which are conviniently 187First, all remaining key => value pairs (all of which are conveniently
183undocumented at the moment) will be used. Then they will be overwritten by 188undocumented at the moment) will be interpreted as configuration
184any values specified in the global default configuration (see the F<aemp> 189data. Then they will be overwritten by any values specified in the global
185utility), then the chain of profiles selected, if any. That means that 190default configuration (see the F<aemp> utility), then the chain of
191profiles chosen by the profile name (and any C<parent> attributes).
192
186the values specified in the profile have highest priority and the values 193That means that the values specified in the profile have highest priority
187specified via C<initialise_node> have lowest priority. 194and the values specified directly via C<configure> have lowest priority,
195and can only be used to specify defaults.
188 196
189If the profile specifies a node ID, then this will become the node ID of 197If the profile specifies a node ID, then this will become the node ID of
190this process. If not, then the profile name will be used as node ID. The 198this process. If not, then the profile name will be used as node ID. The
191special node ID of C<anon/> will be replaced by a random node ID. 199special node ID of C<anon/> will be replaced by a random node ID.
200
201=item step 2, bind listener sockets
192 202
193The next step is to look up the binds in the profile, followed by binding 203The next step is to look up the binds in the profile, followed by binding
194aemp protocol listeners on all binds specified (it is possible and valid 204aemp protocol listeners on all binds specified (it is possible and valid
195to have no binds, meaning that the node cannot be contacted form the 205to have no binds, meaning that the node cannot be contacted form the
196outside. This means the node cannot talk to other nodes that also have no 206outside. This means the node cannot talk to other nodes that also have no
197binds, but it can still talk to all "normal" nodes). 207binds, but it can still talk to all "normal" nodes).
198 208
199If the profile does not specify a binds list, then a default of C<*> is 209If the profile does not specify a binds list, then a default of C<*> is
200used. 210used, meaning the node will bind on a dynamically-assigned port on every
211local IP address it finds.
201 212
213=item step 3, connect to seed nodes
214
202Lastly, the seeds list from the profile is passed to the 215As the last step, the seeds list from the profile is passed to the
203L<AnyEvent::MP::Global> module, which will then use it to keep 216L<AnyEvent::MP::Global> module, which will then use it to keep
204connectivity with at least on of those seed nodes at any point in time. 217connectivity with at least one node at any point in time.
205 218
206Example: become a distributed node listening on the guessed noderef, or 219=back
207the one specified via C<aemp> for the current node. This should be the 220
221Example: become a distributed node using the locla node name as profile.
208most common form of invocation for "daemon"-type nodes. 222This should be the most common form of invocation for "daemon"-type nodes.
209 223
210 initialise_node; 224 configure
211 225
212Example: become an anonymous node. This form is often used for commandline 226Example: become an anonymous node. This form is often used for commandline
213clients. 227clients.
214 228
215 initialise_node "anon/"; 229 configure nodeid => "anon/";
216 230
217Example: become a distributed node. If there is no profile of the given 231Example: configure a node using a profile called seed, which si suitable
218name, or no binds list was specified, resolve C<localhost:4044> and bind 232for a seed node as it binds on all local addresses on a fixed port (4040,
219on the resulting addresses. 233customary for aemp).
220 234
221 initialise_node "localhost:4044"; 235 # use the aemp commandline utility
236 # aemp profile seed nodeid anon/ binds '*:4040'
237
238 # then use it
239 configure profile => "seed";
240
241 # or simply use aemp from the shell again:
242 # aemp run profile seed
243
244 # or provide a nicer-to-remember nodeid
245 # aemp run profile seed nodeid "$(hostname)"
222 246
223=item $SELF 247=item $SELF
224 248
225Contains the current port id while executing C<rcv> callbacks or C<psub> 249Contains the current port id while executing C<rcv> callbacks or C<psub>
226blocks. 250blocks.
348 372
349=cut 373=cut
350 374
351sub rcv($@) { 375sub rcv($@) {
352 my $port = shift; 376 my $port = shift;
353 my ($noderef, $portid) = split /#/, $port, 2; 377 my ($nodeid, $portid) = split /#/, $port, 2;
354 378
355 $NODE{$noderef} == $NODE{""} 379 $NODE{$nodeid} == $NODE{""}
356 or Carp::croak "$port: rcv can only be called on local ports, caught"; 380 or Carp::croak "$port: rcv can only be called on local ports, caught";
357 381
358 while (@_) { 382 while (@_) {
359 if (ref $_[0]) { 383 if (ref $_[0]) {
360 if (my $self = $PORT_DATA{$portid}) { 384 if (my $self = $PORT_DATA{$portid}) {
451 475
452Monitor the given port and do something when the port is killed or 476Monitor the given port and do something when the port is killed or
453messages to it were lost, and optionally return a guard that can be used 477messages to it were lost, and optionally return a guard that can be used
454to stop monitoring again. 478to stop monitoring again.
455 479
480In the first form (callback), the callback is simply called with any
481number of C<@reason> elements (no @reason means that the port was deleted
482"normally"). Note also that I<< the callback B<must> never die >>, so use
483C<eval> if unsure.
484
485In the second form (another port given), the other port (C<$rcvport>)
486will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
487"normal" kils nothing happens, while under all other conditions, the other
488port is killed with the same reason.
489
490The third form (kill self) is the same as the second form, except that
491C<$rvport> defaults to C<$SELF>.
492
493In the last form (message), a message of the form C<@msg, @reason> will be
494C<snd>.
495
496Monitoring-actions are one-shot: once messages are lost (and a monitoring
497alert was raised), they are removed and will not trigger again.
498
499As a rule of thumb, monitoring requests should always monitor a port from
500a local port (or callback). The reason is that kill messages might get
501lost, just like any other message. Another less obvious reason is that
502even monitoring requests can get lost (for example, when the connection
503to the other node goes down permanently). When monitoring a port locally
504these problems do not exist.
505
456C<mon> effectively guarantees that, in the absence of hardware failures, 506C<mon> effectively guarantees that, in the absence of hardware failures,
457after starting the monitor, either all messages sent to the port will 507after starting the monitor, either all messages sent to the port will
458arrive, or the monitoring action will be invoked after possible message 508arrive, or the monitoring action will be invoked after possible message
459loss has been detected. No messages will be lost "in between" (after 509loss has been detected. No messages will be lost "in between" (after
460the first lost message no further messages will be received by the 510the first lost message no further messages will be received by the
461port). After the monitoring action was invoked, further messages might get 511port). After the monitoring action was invoked, further messages might get
462delivered again. 512delivered again.
463 513
464Note that monitoring-actions are one-shot: once messages are lost (and a 514Inter-host-connection timeouts and monitoring depend on the transport
465monitoring alert was raised), they are removed and will not trigger again. 515used. The only transport currently implemented is TCP, and AnyEvent::MP
516relies on TCP to detect node-downs (this can take 10-15 minutes on a
517non-idle connection, and usually around two hours for idle conenctions).
466 518
467In the first form (callback), the callback is simply called with any 519This means that monitoring is good for program errors and cleaning up
468number of C<@reason> elements (no @reason means that the port was deleted 520stuff eventually, but they are no replacement for a timeout when you need
469"normally"). Note also that I<< the callback B<must> never die >>, so use 521to ensure some maximum latency.
470C<eval> if unsure.
471
472In the second form (another port given), the other port (C<$rcvport>)
473will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on
474"normal" kils nothing happens, while under all other conditions, the other
475port is killed with the same reason.
476
477The third form (kill self) is the same as the second form, except that
478C<$rvport> defaults to C<$SELF>.
479
480In the last form (message), a message of the form C<@msg, @reason> will be
481C<snd>.
482
483As a rule of thumb, monitoring requests should always monitor a port from
484a local port (or callback). The reason is that kill messages might get
485lost, just like any other message. Another less obvious reason is that
486even monitoring requests can get lost (for exmaple, when the connection
487to the other node goes down permanently). When monitoring a port locally
488these problems do not exist.
489 522
490Example: call a given callback when C<$port> is killed. 523Example: call a given callback when C<$port> is killed.
491 524
492 mon $port, sub { warn "port died because of <@_>\n" }; 525 mon $port, sub { warn "port died because of <@_>\n" };
493 526
500 mon $port, $self => "restart"; 533 mon $port, $self => "restart";
501 534
502=cut 535=cut
503 536
504sub mon { 537sub mon {
505 my ($noderef, $port) = split /#/, shift, 2; 538 my ($nodeid, $port) = split /#/, shift, 2;
506 539
507 my $node = $NODE{$noderef} || add_node $noderef; 540 my $node = $NODE{$nodeid} || add_node $nodeid;
508 541
509 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,'; 542 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
510 543
511 unless (ref $cb) { 544 unless (ref $cb) {
512 if (@_) { 545 if (@_) {
626 }; 659 };
627 _self_die if $@; 660 _self_die if $@;
628} 661}
629 662
630sub spawn(@) { 663sub spawn(@) {
631 my ($noderef, undef) = split /#/, shift, 2; 664 my ($nodeid, undef) = split /#/, shift, 2;
632 665
633 my $id = "$RUNIQ." . $ID++; 666 my $id = "$RUNIQ." . $ID++;
634 667
635 $_[0] =~ /::/ 668 $_[0] =~ /::/
636 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 669 or Carp::croak "spawn init function must be a fully-qualified name, caught";
637 670
638 snd_to_func $noderef, "AnyEvent::MP::_spawn" => $id, @_; 671 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
639 672
640 "$noderef#$id" 673 "$nodeid#$id"
641} 674}
642 675
643=item after $timeout, @msg 676=item after $timeout, @msg
644 677
645=item after $timeout, $callback 678=item after $timeout, $callback
684 717
685=item * Node IDs are arbitrary strings in AEMP. 718=item * Node IDs are arbitrary strings in AEMP.
686 719
687Erlang relies on special naming and DNS to work everywhere in the same 720Erlang relies on special naming and DNS to work everywhere in the same
688way. AEMP relies on each node somehow knowing its own address(es) (e.g. by 721way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
689configuraiton or DNS), but will otherwise discover other odes itself. 722configuration or DNS), but will otherwise discover other odes itself.
690 723
691=item * Erlang has a "remote ports are like local ports" philosophy, AEMP 724=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
692uses "local ports are like remote ports". 725uses "local ports are like remote ports".
693 726
694The failure modes for local ports are quite different (runtime errors 727The failure modes for local ports are quite different (runtime errors
707 740
708Erlang uses processes that selectively receive messages, and therefore 741Erlang uses processes that selectively receive messages, and therefore
709needs a queue. AEMP is event based, queuing messages would serve no 742needs a queue. AEMP is event based, queuing messages would serve no
710useful purpose. For the same reason the pattern-matching abilities of 743useful purpose. For the same reason the pattern-matching abilities of
711AnyEvent::MP are more limited, as there is little need to be able to 744AnyEvent::MP are more limited, as there is little need to be able to
712filter messages without dequeing them. 745filter messages without dequeuing them.
713 746
714(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 747(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
715 748
716=item * Erlang sends are synchronous, AEMP sends are asynchronous. 749=item * Erlang sends are synchronous, AEMP sends are asynchronous.
717 750

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines