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.69 by root, Sun Aug 30 18:51:49 2009 UTC vs.
Revision 1.86 by root, Wed Sep 9 01:47:01 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
38 mon $port, $otherport # kill otherport on abnormal death 37 mon $port, $otherport # kill otherport on abnormal death
39 mon $port, $otherport, @msg # send message on death 38 mon $port, $otherport, @msg # send message on death
40 39
41=head1 CURRENT STATUS 40=head1 CURRENT STATUS
42 41
42 bin/aemp - stable.
43 AnyEvent::MP - stable API, should work 43 AnyEvent::MP - stable API, should work.
44 AnyEvent::MP::Intro - outdated 44 AnyEvent::MP::Intro - explains most concepts.
45 AnyEvent::MP::Kernel - mostly stable 45 AnyEvent::MP::Kernel - mostly stable.
46 AnyEvent::MP::Global - mostly stable 46 AnyEvent::MP::Global - stable but incomplete, protocol not yet final.
47 AnyEvent::MP::Node - mostly stable, but internal anyways
48 AnyEvent::MP::Transport - mostly stable, but internal anyways
49 47
50 stay tuned. 48stay tuned.
51 49
52=head1 DESCRIPTION 50=head1 DESCRIPTION
53 51
54This module (-family) implements a simple message passing framework. 52This module (-family) implements a simple message passing framework.
55 53
57on the same or other hosts, and you can supervise entities remotely. 55on the same or other hosts, and you can supervise entities remotely.
58 56
59For 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>
60manual page and the examples under F<eg/>. 58manual page and the examples under F<eg/>.
61 59
62At the moment, this module family is a bit underdocumented.
63
64=head1 CONCEPTS 60=head1 CONCEPTS
65 61
66=over 4 62=over 4
67 63
68=item port 64=item port
69 65
70A 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).
71 68
72Ports 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
73some messages. Messages send to ports will not be queued, regardless of 70some messages. Messages send to ports will not be queued, regardless of
74anything was listening for them or not. 71anything was listening for them or not.
75 72
86 83
87Nodes are either public (have one or more listening ports) or private 84Nodes are either public (have one or more listening ports) or private
88(no listening ports). Private nodes cannot talk to other private nodes 85(no listening ports). Private nodes cannot talk to other private nodes
89currently. 86currently.
90 87
91=item node ID - C<[a-za-Z0-9_\-.:]+> 88=item node ID - C<[A-Z_][a-zA-Z0-9_\-.:]*>
92 89
93A node ID is a string that uniquely identifies the node within a 90A node ID is a string that uniquely identifies the node within a
94network. Depending on the configuration used, node IDs can look like a 91network. Depending on the configuration used, node IDs can look like a
95hostname, a hostname and a port, or a random string. AnyEvent::MP itself 92hostname, a hostname and a port, or a random string. AnyEvent::MP itself
96doesn't interpret node IDs in any way. 93doesn't interpret node IDs in any way.
100Nodes can only talk to each other by creating some kind of connection to 97Nodes can only talk to each other by creating some kind of connection to
101each other. To do this, nodes should listen on one or more local transport 98each other. To do this, nodes should listen on one or more local transport
102endpoints - binds. Currently, only standard C<ip:port> specifications can 99endpoints - binds. Currently, only standard C<ip:port> specifications can
103be used, which specify TCP ports to listen on. 100be used, which specify TCP ports to listen on.
104 101
105=item seeds - C<host:port> 102=item seed nodes
106 103
107When a node starts, it knows nothing about the network. To teach the node 104When a node starts, it knows nothing about the network. To teach the node
108about the network it first has to contact some other node within the 105about the network it first has to contact some other node within the
109network. This node is called a seed. 106network. This node is called a seed.
110 107
111Seeds are transport endpoint(s) of as many nodes as one wants. Those nodes 108Apart from the fact that other nodes know them as seed nodes and they have
109to have fixed listening addresses, seed nodes are perfectly normal nodes -
110any node can function as a seed node for others.
111
112In addition to discovering the network, seed nodes are also used to
113maintain the network and to connect nodes that otherwise would have
114trouble connecting. They form the backbone of an AnyEvent::MP network.
115
112are expected to be long-running, and at least one of those should always 116Seed nodes are expected to be long-running, and at least one seed node
113be available. When nodes run out of connections (e.g. due to a network 117should always be available. They should also be relatively responsive - a
114error), they try to re-establish connections to some seednodes again to 118seed node that blocks for long periods will slow down everybody else.
115join the network.
116 119
117Apart from being sued for seeding, seednodes are not special in any way - 120=item seeds - C<host:port>
118every public node can be a seednode. 121
122Seeds are transport endpoint(s) (usually a hostname/IP address and a
123TCP port) of nodes thta should be used as seed nodes.
124
125The nodes listening on those endpoints are expected to be long-running,
126and at least one of those should always be available. When nodes run out
127of connections (e.g. due to a network error), they try to re-establish
128connections to some seednodes again to join the network.
119 129
120=back 130=back
121 131
122=head1 VARIABLES/FUNCTIONS 132=head1 VARIABLES/FUNCTIONS
123 133
139 149
140our $VERSION = $AnyEvent::MP::Kernel::VERSION; 150our $VERSION = $AnyEvent::MP::Kernel::VERSION;
141 151
142our @EXPORT = qw( 152our @EXPORT = qw(
143 NODE $NODE *SELF node_of after 153 NODE $NODE *SELF node_of after
144 initialise_node 154 configure
145 snd rcv mon mon_guard kil reg psub spawn 155 snd rcv mon mon_guard kil reg psub spawn
146 port 156 port
147); 157);
148 158
149our $SELF; 159our $SELF;
156 166
157=item $thisnode = NODE / $NODE 167=item $thisnode = NODE / $NODE
158 168
159The C<NODE> function returns, and the C<$NODE> variable contains, the node 169The C<NODE> function returns, and the C<$NODE> variable contains, the node
160ID of the node running in the current process. This value is initialised by 170ID of the node running in the current process. This value is initialised by
161a call to C<initialise_node>. 171a call to C<configure>.
162 172
163=item $nodeid = node_of $port 173=item $nodeid = node_of $port
164 174
165Extracts and returns the node ID from a port ID or a node ID. 175Extracts and returns the node ID from a port ID or a node ID.
166 176
167=item initialise_node $profile_name, key => value... 177=item configure $profile, key => value...
178
179=item configure key => value...
168 180
169Before a node can talk to other nodes on the network (i.e. enter 181Before a node can talk to other nodes on the network (i.e. enter
170"distributed mode") it has to initialise itself - the minimum a node needs 182"distributed mode") it has to configure itself - the minimum a node needs
171to know is its own name, and optionally it should know the addresses of 183to know is its own name, and optionally it should know the addresses of
172some other nodes in the network to discover other nodes. 184some other nodes in the network to discover other nodes.
173 185
174This function initialises a node - it must be called exactly once (or 186This function configures a node - it must be called exactly once (or
175never) before calling other AnyEvent::MP functions. 187never) before calling other AnyEvent::MP functions.
176 188
177The first argument is a profile name. If it is C<undef> or missing, then 189=over 4
178the current nodename will be used instead (i.e. F<uname -n>).
179 190
191=item step 1, gathering configuration from profiles
192
180The function first looks up the profile in the aemp configuration (see the 193The function first looks up a profile in the aemp configuration (see the
181L<aemp> commandline utility). the profile is calculated as follows: 194L<aemp> commandline utility). The profile name can be specified via the
195named C<profile> parameter or can simply be the first parameter). If it is
196missing, then the nodename (F<uname -n>) will be used as profile name.
182 197
183First, all remaining key => value pairs will be used. Then they will be 198The profile data is then gathered as follows:
184overwritten by any values specified in the global default configuration 199
185(see the F<aemp> utility), then the chain of profiles selected, if 200First, all remaining key => value pairs (all of which are conveniently
201undocumented at the moment) will be interpreted as configuration
202data. Then they will be overwritten by any values specified in the global
203default configuration (see the F<aemp> utility), then the chain of
204profiles chosen by the profile name (and any C<parent> attributes).
205
186any. That means that the values specified in the profile have highest 206That means that the values specified in the profile have highest priority
187priority and the values specified via C<initialise_node> have lowest 207and the values specified directly via C<configure> have lowest priority,
188priority. 208and can only be used to specify defaults.
189 209
190If the profile specifies a node ID, then this will become the node ID of 210If the profile specifies a node ID, then this will become the node ID of
191this process. If not, then the profile name will be used as node ID. The 211this process. If not, then the profile name will be used as node ID. The
192special node ID of C<anon/> will be replaced by a random node ID. 212special node ID of C<anon/> will be replaced by a random node ID.
213
214=item step 2, bind listener sockets
193 215
194The next step is to look up the binds in the profile, followed by binding 216The next step is to look up the binds in the profile, followed by binding
195aemp protocol listeners on all binds specified (it is possible and valid 217aemp protocol listeners on all binds specified (it is possible and valid
196to have no binds, meaning that the node cannot be contacted form the 218to have no binds, meaning that the node cannot be contacted form the
197outside. This means the node cannot talk to other nodes that also have no 219outside. This means the node cannot talk to other nodes that also have no
198binds, but it can still talk to all "normal" nodes). 220binds, but it can still talk to all "normal" nodes).
199 221
200If the profile does not specify a binds list, then the node ID will be 222If the profile does not specify a binds list, then a default of C<*> is
201treated as if it were of the form C<host:port>, which will be resolved and 223used, meaning the node will bind on a dynamically-assigned port on every
202used as binds list. 224local IP address it finds.
203 225
226=item step 3, connect to seed nodes
227
204Lastly, the seeds list from the profile is passed to the 228As the last step, the seeds list from the profile is passed to the
205L<AnyEvent::MP::Global> module, which will then use it to keep 229L<AnyEvent::MP::Global> module, which will then use it to keep
206connectivity with at least on of those seed nodes at any point in time. 230connectivity with at least one node at any point in time.
207 231
208Example: become a distributed node listening on the guessed noderef, or 232=back
209the one specified via C<aemp> for the current node. This should be the 233
234Example: become a distributed node using the locla node name as profile.
210most common form of invocation for "daemon"-type nodes. 235This should be the most common form of invocation for "daemon"-type nodes.
211 236
212 initialise_node; 237 configure
213 238
214Example: become an anonymous node. This form is often used for commandline 239Example: become an anonymous node. This form is often used for commandline
215clients. 240clients.
216 241
217 initialise_node "anon/"; 242 configure nodeid => "anon/";
218 243
219Example: become a distributed node. If there is no profile of the given 244Example: configure a node using a profile called seed, which si suitable
220name, or no binds list was specified, resolve C<localhost:4044> and bind 245for a seed node as it binds on all local addresses on a fixed port (4040,
221on the resulting addresses. 246customary for aemp).
222 247
223 initialise_node "localhost:4044"; 248 # use the aemp commandline utility
249 # aemp profile seed nodeid anon/ binds '*:4040'
250
251 # then use it
252 configure profile => "seed";
253
254 # or simply use aemp from the shell again:
255 # aemp run profile seed
256
257 # or provide a nicer-to-remember nodeid
258 # aemp run profile seed nodeid "$(hostname)"
224 259
225=item $SELF 260=item $SELF
226 261
227Contains the current port id while executing C<rcv> callbacks or C<psub> 262Contains the current port id while executing C<rcv> callbacks or C<psub>
228blocks. 263blocks.
350 385
351=cut 386=cut
352 387
353sub rcv($@) { 388sub rcv($@) {
354 my $port = shift; 389 my $port = shift;
355 my ($noderef, $portid) = split /#/, $port, 2; 390 my ($nodeid, $portid) = split /#/, $port, 2;
356 391
357 $NODE{$noderef} == $NODE{""} 392 $NODE{$nodeid} == $NODE{""}
358 or Carp::croak "$port: rcv can only be called on local ports, caught"; 393 or Carp::croak "$port: rcv can only be called on local ports, caught";
359 394
360 while (@_) { 395 while (@_) {
361 if (ref $_[0]) { 396 if (ref $_[0]) {
362 if (my $self = $PORT_DATA{$portid}) { 397 if (my $self = $PORT_DATA{$portid}) {
453 488
454Monitor the given port and do something when the port is killed or 489Monitor the given port and do something when the port is killed or
455messages to it were lost, and optionally return a guard that can be used 490messages to it were lost, and optionally return a guard that can be used
456to stop monitoring again. 491to stop monitoring again.
457 492
493In the first form (callback), the callback is simply called with any
494number of C<@reason> elements (no @reason means that the port was deleted
495"normally"). Note also that I<< the callback B<must> never die >>, so use
496C<eval> if unsure.
497
498In the second form (another port given), the other port (C<$rcvport>)
499will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
500"normal" kils nothing happens, while under all other conditions, the other
501port is killed with the same reason.
502
503The third form (kill self) is the same as the second form, except that
504C<$rvport> defaults to C<$SELF>.
505
506In the last form (message), a message of the form C<@msg, @reason> will be
507C<snd>.
508
509Monitoring-actions are one-shot: once messages are lost (and a monitoring
510alert was raised), they are removed and will not trigger again.
511
512As a rule of thumb, monitoring requests should always monitor a port from
513a local port (or callback). The reason is that kill messages might get
514lost, just like any other message. Another less obvious reason is that
515even monitoring requests can get lost (for example, when the connection
516to the other node goes down permanently). When monitoring a port locally
517these problems do not exist.
518
458C<mon> effectively guarantees that, in the absence of hardware failures, 519C<mon> effectively guarantees that, in the absence of hardware failures,
459after starting the monitor, either all messages sent to the port will 520after starting the monitor, either all messages sent to the port will
460arrive, or the monitoring action will be invoked after possible message 521arrive, or the monitoring action will be invoked after possible message
461loss has been detected. No messages will be lost "in between" (after 522loss has been detected. No messages will be lost "in between" (after
462the first lost message no further messages will be received by the 523the first lost message no further messages will be received by the
463port). After the monitoring action was invoked, further messages might get 524port). After the monitoring action was invoked, further messages might get
464delivered again. 525delivered again.
465 526
466Note that monitoring-actions are one-shot: once messages are lost (and a 527Inter-host-connection timeouts and monitoring depend on the transport
467monitoring alert was raised), they are removed and will not trigger again. 528used. The only transport currently implemented is TCP, and AnyEvent::MP
529relies on TCP to detect node-downs (this can take 10-15 minutes on a
530non-idle connection, and usually around two hours for idle conenctions).
468 531
469In the first form (callback), the callback is simply called with any 532This means that monitoring is good for program errors and cleaning up
470number of C<@reason> elements (no @reason means that the port was deleted 533stuff eventually, but they are no replacement for a timeout when you need
471"normally"). Note also that I<< the callback B<must> never die >>, so use 534to ensure some maximum latency.
472C<eval> if unsure.
473
474In the second form (another port given), the other port (C<$rcvport>)
475will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on
476"normal" kils nothing happens, while under all other conditions, the other
477port is killed with the same reason.
478
479The third form (kill self) is the same as the second form, except that
480C<$rvport> defaults to C<$SELF>.
481
482In the last form (message), a message of the form C<@msg, @reason> will be
483C<snd>.
484
485As a rule of thumb, monitoring requests should always monitor a port from
486a local port (or callback). The reason is that kill messages might get
487lost, just like any other message. Another less obvious reason is that
488even monitoring requests can get lost (for exmaple, when the connection
489to the other node goes down permanently). When monitoring a port locally
490these problems do not exist.
491 535
492Example: call a given callback when C<$port> is killed. 536Example: call a given callback when C<$port> is killed.
493 537
494 mon $port, sub { warn "port died because of <@_>\n" }; 538 mon $port, sub { warn "port died because of <@_>\n" };
495 539
502 mon $port, $self => "restart"; 546 mon $port, $self => "restart";
503 547
504=cut 548=cut
505 549
506sub mon { 550sub mon {
507 my ($noderef, $port) = split /#/, shift, 2; 551 my ($nodeid, $port) = split /#/, shift, 2;
508 552
509 my $node = $NODE{$noderef} || add_node $noderef; 553 my $node = $NODE{$nodeid} || add_node $nodeid;
510 554
511 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,'; 555 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
512 556
513 unless (ref $cb) { 557 unless (ref $cb) {
514 if (@_) { 558 if (@_) {
590the package, then the package above the package and so on (e.g. 634the package, then the package above the package and so on (e.g.
591C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function 635C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
592exists or it runs out of package names. 636exists or it runs out of package names.
593 637
594The init function is then called with the newly-created port as context 638The init function is then called with the newly-created port as context
595object (C<$SELF>) and the C<@initdata> values as arguments. 639object (C<$SELF>) and the C<@initdata> values as arguments. It I<must>
640call one of the C<rcv> functions to set callbacks on C<$SELF>, otherwise
641the port might not get created.
596 642
597A common idiom is to pass a local port, immediately monitor the spawned 643A common idiom is to pass a local port, immediately monitor the spawned
598port, and in the remote init function, immediately monitor the passed 644port, and in the remote init function, immediately monitor the passed
599local port. This two-way monitoring ensures that both ports get cleaned up 645local port. This two-way monitoring ensures that both ports get cleaned up
600when there is a problem. 646when there is a problem.
601 647
648C<spawn> guarantees that the C<$initfunc> has no visible effects on the
649caller before C<spawn> returns (by delaying invocation when spawn is
650called for the local node).
651
602Example: spawn a chat server port on C<$othernode>. 652Example: spawn a chat server port on C<$othernode>.
603 653
604 # this node, executed from within a port context: 654 # this node, executed from within a port context:
605 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF; 655 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
606 mon $server; 656 mon $server;
620 670
621sub _spawn { 671sub _spawn {
622 my $port = shift; 672 my $port = shift;
623 my $init = shift; 673 my $init = shift;
624 674
675 # rcv will create the actual port
625 local $SELF = "$NODE#$port"; 676 local $SELF = "$NODE#$port";
626 eval { 677 eval {
627 &{ load_func $init } 678 &{ load_func $init }
628 }; 679 };
629 _self_die if $@; 680 _self_die if $@;
630} 681}
631 682
632sub spawn(@) { 683sub spawn(@) {
633 my ($noderef, undef) = split /#/, shift, 2; 684 my ($nodeid, undef) = split /#/, shift, 2;
634 685
635 my $id = "$RUNIQ." . $ID++; 686 my $id = "$RUNIQ." . $ID++;
636 687
637 $_[0] =~ /::/ 688 $_[0] =~ /::/
638 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 689 or Carp::croak "spawn init function must be a fully-qualified name, caught";
639 690
640 snd_to_func $noderef, "AnyEvent::MP::_spawn" => $id, @_; 691 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
641 692
642 "$noderef#$id" 693 "$nodeid#$id"
643} 694}
644 695
645=item after $timeout, @msg 696=item after $timeout, @msg
646 697
647=item after $timeout, $callback 698=item after $timeout, $callback
686 737
687=item * Node IDs are arbitrary strings in AEMP. 738=item * Node IDs are arbitrary strings in AEMP.
688 739
689Erlang relies on special naming and DNS to work everywhere in the same 740Erlang relies on special naming and DNS to work everywhere in the same
690way. AEMP relies on each node somehow knowing its own address(es) (e.g. by 741way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
691configuraiton or DNS), but will otherwise discover other odes itself. 742configuration or DNS), but will otherwise discover other odes itself.
692 743
693=item * Erlang has a "remote ports are like local ports" philosophy, AEMP 744=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
694uses "local ports are like remote ports". 745uses "local ports are like remote ports".
695 746
696The failure modes for local ports are quite different (runtime errors 747The failure modes for local ports are quite different (runtime errors
709 760
710Erlang uses processes that selectively receive messages, and therefore 761Erlang uses processes that selectively receive messages, and therefore
711needs a queue. AEMP is event based, queuing messages would serve no 762needs a queue. AEMP is event based, queuing messages would serve no
712useful purpose. For the same reason the pattern-matching abilities of 763useful purpose. For the same reason the pattern-matching abilities of
713AnyEvent::MP are more limited, as there is little need to be able to 764AnyEvent::MP are more limited, as there is little need to be able to
714filter messages without dequeing them. 765filter messages without dequeuing them.
715 766
716(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 767(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
717 768
718=item * Erlang sends are synchronous, AEMP sends are asynchronous. 769=item * Erlang sends are synchronous, AEMP sends are asynchronous.
719 770
825L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 876L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
826 877
827L<AnyEvent::MP::Global> - network maintainance and port groups, to find 878L<AnyEvent::MP::Global> - network maintainance and port groups, to find
828your applications. 879your applications.
829 880
881L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
882all nodes.
883
830L<AnyEvent>. 884L<AnyEvent>.
831 885
832=head1 AUTHOR 886=head1 AUTHOR
833 887
834 Marc Lehmann <schmorp@schmorp.de> 888 Marc Lehmann <schmorp@schmorp.de>

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines