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.74 by root, Mon Aug 31 11:11:27 2009 UTC vs.
Revision 1.95 by root, Wed Sep 23 11:57:16 2009 UTC

1=head1 NAME 1=head1 NAME
2 2
3AnyEvent::MP - multi-processing/message-passing framework 3AnyEvent::MP - erlang-style multi-processing/message-passing framework
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 configure; 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;
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
37 mon $port, $cb->(@msg) # callback is invoked on death 36 mon $localport, $cb->(@msg) # callback is invoked on death
38 mon $port, $otherport # kill otherport on abnormal death 37 mon $localport, $otherport # kill otherport on abnormal death
39 mon $port, $otherport, @msg # send message on death 38 mon $localport, $otherport, @msg # send message on death
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 API.
47 AnyEvent::MP::Global - stable API, protocol not yet final. 46 AnyEvent::MP::Global - stable API.
48
49 stay tuned.
50 47
51=head1 DESCRIPTION 48=head1 DESCRIPTION
52 49
53This module (-family) implements a simple message passing framework. 50This module (-family) implements a simple message passing framework.
54 51
56on the same or other hosts, and you can supervise entities remotely. 53on the same or other hosts, and you can supervise entities remotely.
57 54
58For an introduction to this module family, see the L<AnyEvent::MP::Intro> 55For an introduction to this module family, see the L<AnyEvent::MP::Intro>
59manual page and the examples under F<eg/>. 56manual page and the examples under F<eg/>.
60 57
61At the moment, this module family is a bit underdocumented.
62
63=head1 CONCEPTS 58=head1 CONCEPTS
64 59
65=over 4 60=over 4
66 61
67=item port 62=item port
68 63
69A port is something you can send messages to (with the C<snd> function). 64Not to be confused with a TCP port, a "port" is something you can send
65messages to (with the C<snd> function).
70 66
71Ports allow you to register C<rcv> handlers that can match all or just 67Ports 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 68some messages. Messages send to ports will not be queued, regardless of
73anything was listening for them or not. 69anything was listening for them or not.
74 70
85 81
86Nodes are either public (have one or more listening ports) or private 82Nodes are either public (have one or more listening ports) or private
87(no listening ports). Private nodes cannot talk to other private nodes 83(no listening ports). Private nodes cannot talk to other private nodes
88currently. 84currently.
89 85
90=item node ID - C<[a-za-Z0-9_\-.:]+> 86=item node ID - C<[A-Z_][a-zA-Z0-9_\-.:]*>
91 87
92A node ID is a string that uniquely identifies the node within a 88A node ID is a string that uniquely identifies the node within a
93network. Depending on the configuration used, node IDs can look like a 89network. Depending on the configuration used, node IDs can look like a
94hostname, a hostname and a port, or a random string. AnyEvent::MP itself 90hostname, a hostname and a port, or a random string. AnyEvent::MP itself
95doesn't interpret node IDs in any way. 91doesn't interpret node IDs in any way.
99Nodes can only talk to each other by creating some kind of connection to 95Nodes can only talk to each other by creating some kind of connection to
100each other. To do this, nodes should listen on one or more local transport 96each other. To do this, nodes should listen on one or more local transport
101endpoints - binds. Currently, only standard C<ip:port> specifications can 97endpoints - binds. Currently, only standard C<ip:port> specifications can
102be used, which specify TCP ports to listen on. 98be used, which specify TCP ports to listen on.
103 99
104=item seeds - C<host:port> 100=item seed nodes
105 101
106When a node starts, it knows nothing about the network. To teach the node 102When a node starts, it knows nothing about the network. To teach the node
107about the network it first has to contact some other node within the 103about the network it first has to contact some other node within the
108network. This node is called a seed. 104network. This node is called a seed.
109 105
110Seeds are transport endpoint(s) of as many nodes as one wants. Those nodes 106Apart from the fact that other nodes know them as seed nodes and they have
107to have fixed listening addresses, seed nodes are perfectly normal nodes -
108any node can function as a seed node for others.
109
110In addition to discovering the network, seed nodes are also used to
111maintain the network and to connect nodes that otherwise would have
112trouble connecting. They form the backbone of an AnyEvent::MP network.
113
111are expected to be long-running, and at least one of those should always 114Seed nodes are expected to be long-running, and at least one seed node
112be available. When nodes run out of connections (e.g. due to a network 115should always be available. They should also be relatively responsive - a
113error), they try to re-establish connections to some seednodes again to 116seed node that blocks for long periods will slow down everybody else.
114join the network.
115 117
116Apart from being sued for seeding, seednodes are not special in any way - 118=item seeds - C<host:port>
117every public node can be a seednode. 119
120Seeds are transport endpoint(s) (usually a hostname/IP address and a
121TCP port) of nodes thta should be used as seed nodes.
122
123The nodes listening on those endpoints are expected to be long-running,
124and at least one of those should always be available. When nodes run out
125of connections (e.g. due to a network error), they try to re-establish
126connections to some seednodes again to join the network.
118 127
119=back 128=back
120 129
121=head1 VARIABLES/FUNCTIONS 130=head1 VARIABLES/FUNCTIONS
122 131
139our $VERSION = $AnyEvent::MP::Kernel::VERSION; 148our $VERSION = $AnyEvent::MP::Kernel::VERSION;
140 149
141our @EXPORT = qw( 150our @EXPORT = qw(
142 NODE $NODE *SELF node_of after 151 NODE $NODE *SELF node_of after
143 configure 152 configure
144 snd rcv mon mon_guard kil reg psub spawn 153 snd rcv mon mon_guard kil psub spawn cal
145 port 154 port
146); 155);
147 156
148our $SELF; 157our $SELF;
149 158
161 170
162=item $nodeid = node_of $port 171=item $nodeid = node_of $port
163 172
164Extracts and returns the node ID from a port ID or a node ID. 173Extracts and returns the node ID from a port ID or a node ID.
165 174
175=item configure $profile, key => value...
176
166=item configure key => value... 177=item configure key => value...
167 178
168Before a node can talk to other nodes on the network (i.e. enter 179Before a node can talk to other nodes on the network (i.e. enter
169"distributed mode") it has to configure itself - the minimum a node needs 180"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 181to know is its own name, and optionally it should know the addresses of
177 188
178=item step 1, gathering configuration from profiles 189=item step 1, gathering configuration from profiles
179 190
180The function first looks up a profile in the aemp configuration (see the 191The function first looks up a profile in the aemp configuration (see the
181L<aemp> commandline utility). The profile name can be specified via the 192L<aemp> commandline utility). The profile name can be specified via the
182named C<profile> parameter. If it is missing, then the nodename (F<uname 193named C<profile> parameter or can simply be the first parameter). If it is
183-n>) will be used as profile name. 194missing, then the nodename (F<uname -n>) will be used as profile name.
184 195
185The profile data is then gathered as follows: 196The profile data is then gathered as follows:
186 197
187First, all remaining key => value pairs (all of which are conviniently 198First, all remaining key => value pairs (all of which are conveniently
188undocumented at the moment) will be interpreted as configuration 199undocumented at the moment) will be interpreted as configuration
189data. Then they will be overwritten by any values specified in the global 200data. Then they will be overwritten by any values specified in the global
190default configuration (see the F<aemp> utility), then the chain of 201default configuration (see the F<aemp> utility), then the chain of
191profiles chosen by the profile name (and any C<parent> attributes). 202profiles chosen by the profile name (and any C<parent> attributes).
192 203
216L<AnyEvent::MP::Global> module, which will then use it to keep 227L<AnyEvent::MP::Global> module, which will then use it to keep
217connectivity with at least one node at any point in time. 228connectivity with at least one node at any point in time.
218 229
219=back 230=back
220 231
221Example: become a distributed node using the locla node name as profile. 232Example: become a distributed node using the local node name as profile.
222This should be the most common form of invocation for "daemon"-type nodes. 233This should be the most common form of invocation for "daemon"-type nodes.
223 234
224 configure 235 configure
225 236
226Example: become an anonymous node. This form is often used for commandline 237Example: become an anonymous node. This form is often used for commandline
372 383
373=cut 384=cut
374 385
375sub rcv($@) { 386sub rcv($@) {
376 my $port = shift; 387 my $port = shift;
377 my ($noderef, $portid) = split /#/, $port, 2; 388 my ($nodeid, $portid) = split /#/, $port, 2;
378 389
379 $NODE{$noderef} == $NODE{""} 390 $NODE{$nodeid} == $NODE{""}
380 or Carp::croak "$port: rcv can only be called on local ports, caught"; 391 or Carp::croak "$port: rcv can only be called on local ports, caught";
381 392
382 while (@_) { 393 while (@_) {
383 if (ref $_[0]) { 394 if (ref $_[0]) {
384 if (my $self = $PORT_DATA{$portid}) { 395 if (my $self = $PORT_DATA{$portid}) {
475 486
476Monitor the given port and do something when the port is killed or 487Monitor the given port and do something when the port is killed or
477messages to it were lost, and optionally return a guard that can be used 488messages to it were lost, and optionally return a guard that can be used
478to stop monitoring again. 489to stop monitoring again.
479 490
491In the first form (callback), the callback is simply called with any
492number of C<@reason> elements (no @reason means that the port was deleted
493"normally"). Note also that I<< the callback B<must> never die >>, so use
494C<eval> if unsure.
495
496In the second form (another port given), the other port (C<$rcvport>)
497will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
498"normal" kils nothing happens, while under all other conditions, the other
499port is killed with the same reason.
500
501The third form (kill self) is the same as the second form, except that
502C<$rvport> defaults to C<$SELF>.
503
504In the last form (message), a message of the form C<@msg, @reason> will be
505C<snd>.
506
507Monitoring-actions are one-shot: once messages are lost (and a monitoring
508alert was raised), they are removed and will not trigger again.
509
510As a rule of thumb, monitoring requests should always monitor a port from
511a local port (or callback). The reason is that kill messages might get
512lost, just like any other message. Another less obvious reason is that
513even monitoring requests can get lost (for example, when the connection
514to the other node goes down permanently). When monitoring a port locally
515these problems do not exist.
516
480C<mon> effectively guarantees that, in the absence of hardware failures, 517C<mon> effectively guarantees that, in the absence of hardware failures,
481after starting the monitor, either all messages sent to the port will 518after starting the monitor, either all messages sent to the port will
482arrive, or the monitoring action will be invoked after possible message 519arrive, or the monitoring action will be invoked after possible message
483loss has been detected. No messages will be lost "in between" (after 520loss has been detected. No messages will be lost "in between" (after
484the first lost message no further messages will be received by the 521the first lost message no further messages will be received by the
485port). After the monitoring action was invoked, further messages might get 522port). After the monitoring action was invoked, further messages might get
486delivered again. 523delivered again.
487 524
488Note that monitoring-actions are one-shot: once messages are lost (and a 525Inter-host-connection timeouts and monitoring depend on the transport
489monitoring alert was raised), they are removed and will not trigger again. 526used. The only transport currently implemented is TCP, and AnyEvent::MP
527relies on TCP to detect node-downs (this can take 10-15 minutes on a
528non-idle connection, and usually around two hours for idle conenctions).
490 529
491In the first form (callback), the callback is simply called with any 530This means that monitoring is good for program errors and cleaning up
492number of C<@reason> elements (no @reason means that the port was deleted 531stuff eventually, but they are no replacement for a timeout when you need
493"normally"). Note also that I<< the callback B<must> never die >>, so use 532to ensure some maximum latency.
494C<eval> if unsure.
495
496In the second form (another port given), the other port (C<$rcvport>)
497will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on
498"normal" kils nothing happens, while under all other conditions, the other
499port is killed with the same reason.
500
501The third form (kill self) is the same as the second form, except that
502C<$rvport> defaults to C<$SELF>.
503
504In the last form (message), a message of the form C<@msg, @reason> will be
505C<snd>.
506
507As a rule of thumb, monitoring requests should always monitor a port from
508a local port (or callback). The reason is that kill messages might get
509lost, just like any other message. Another less obvious reason is that
510even monitoring requests can get lost (for exmaple, when the connection
511to the other node goes down permanently). When monitoring a port locally
512these problems do not exist.
513 533
514Example: call a given callback when C<$port> is killed. 534Example: call a given callback when C<$port> is killed.
515 535
516 mon $port, sub { warn "port died because of <@_>\n" }; 536 mon $port, sub { warn "port died because of <@_>\n" };
517 537
524 mon $port, $self => "restart"; 544 mon $port, $self => "restart";
525 545
526=cut 546=cut
527 547
528sub mon { 548sub mon {
529 my ($noderef, $port) = split /#/, shift, 2; 549 my ($nodeid, $port) = split /#/, shift, 2;
530 550
531 my $node = $NODE{$noderef} || add_node $noderef; 551 my $node = $NODE{$nodeid} || add_node $nodeid;
532 552
533 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,'; 553 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
534 554
535 unless (ref $cb) { 555 unless (ref $cb) {
536 if (@_) { 556 if (@_) {
545 } 565 }
546 566
547 $node->monitor ($port, $cb); 567 $node->monitor ($port, $cb);
548 568
549 defined wantarray 569 defined wantarray
550 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) } 570 and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) })
551} 571}
552 572
553=item $guard = mon_guard $port, $ref, $ref... 573=item $guard = mon_guard $port, $ref, $ref...
554 574
555Monitors the given C<$port> and keeps the passed references. When the port 575Monitors the given C<$port> and keeps the passed references. When the port
612the package, then the package above the package and so on (e.g. 632the package, then the package above the package and so on (e.g.
613C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function 633C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
614exists or it runs out of package names. 634exists or it runs out of package names.
615 635
616The init function is then called with the newly-created port as context 636The init function is then called with the newly-created port as context
617object (C<$SELF>) and the C<@initdata> values as arguments. 637object (C<$SELF>) and the C<@initdata> values as arguments. It I<must>
638call one of the C<rcv> functions to set callbacks on C<$SELF>, otherwise
639the port might not get created.
618 640
619A common idiom is to pass a local port, immediately monitor the spawned 641A common idiom is to pass a local port, immediately monitor the spawned
620port, and in the remote init function, immediately monitor the passed 642port, and in the remote init function, immediately monitor the passed
621local port. This two-way monitoring ensures that both ports get cleaned up 643local port. This two-way monitoring ensures that both ports get cleaned up
622when there is a problem. 644when there is a problem.
623 645
646C<spawn> guarantees that the C<$initfunc> has no visible effects on the
647caller before C<spawn> returns (by delaying invocation when spawn is
648called for the local node).
649
624Example: spawn a chat server port on C<$othernode>. 650Example: spawn a chat server port on C<$othernode>.
625 651
626 # this node, executed from within a port context: 652 # this node, executed from within a port context:
627 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF; 653 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
628 mon $server; 654 mon $server;
642 668
643sub _spawn { 669sub _spawn {
644 my $port = shift; 670 my $port = shift;
645 my $init = shift; 671 my $init = shift;
646 672
673 # rcv will create the actual port
647 local $SELF = "$NODE#$port"; 674 local $SELF = "$NODE#$port";
648 eval { 675 eval {
649 &{ load_func $init } 676 &{ load_func $init }
650 }; 677 };
651 _self_die if $@; 678 _self_die if $@;
652} 679}
653 680
654sub spawn(@) { 681sub spawn(@) {
655 my ($noderef, undef) = split /#/, shift, 2; 682 my ($nodeid, undef) = split /#/, shift, 2;
656 683
657 my $id = "$RUNIQ." . $ID++; 684 my $id = "$RUNIQ." . $ID++;
658 685
659 $_[0] =~ /::/ 686 $_[0] =~ /::/
660 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 687 or Carp::croak "spawn init function must be a fully-qualified name, caught";
661 688
662 snd_to_func $noderef, "AnyEvent::MP::_spawn" => $id, @_; 689 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
663 690
664 "$noderef#$id" 691 "$nodeid#$id"
665} 692}
666 693
667=item after $timeout, @msg 694=item after $timeout, @msg
668 695
669=item after $timeout, $callback 696=item after $timeout, $callback
686 ? $action[0]() 713 ? $action[0]()
687 : snd @action; 714 : snd @action;
688 }; 715 };
689} 716}
690 717
718=item cal $port, @msg, $callback[, $timeout]
719
720A simple form of RPC - sends a message to the given C<$port> with the
721given contents (C<@msg>), but adds a reply port to the message.
722
723The reply port is created temporarily just for the purpose of receiving
724the reply, and will be C<kil>ed when no longer needed.
725
726A reply message sent to the port is passed to the C<$callback> as-is.
727
728If an optional time-out (in seconds) is given and it is not C<undef>,
729then the callback will be called without any arguments after the time-out
730elapsed and the port is C<kil>ed.
731
732If no time-out is given, then the local port will monitor the remote port
733instead, so it eventually gets cleaned-up.
734
735Currently this function returns the temporary port, but this "feature"
736might go in future versions unless you can make a convincing case that
737this is indeed useful for something.
738
739=cut
740
741sub cal(@) {
742 my $timeout = ref $_[-1] ? undef : pop;
743 my $cb = pop;
744
745 my $port = port {
746 undef $timeout;
747 kil $SELF;
748 &$cb;
749 };
750
751 if (defined $timeout) {
752 $timeout = AE::timer $timeout, 0, sub {
753 undef $timeout;
754 kil $port;
755 $cb->();
756 };
757 } else {
758 mon $_[0], sub {
759 kil $port;
760 $cb->();
761 };
762 }
763
764 push @_, $port;
765 &snd;
766
767 $port
768}
769
691=back 770=back
692 771
693=head1 AnyEvent::MP vs. Distributed Erlang 772=head1 AnyEvent::MP vs. Distributed Erlang
694 773
695AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 774AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
696== aemp node, Erlang process == aemp port), so many of the documents and 775== aemp node, Erlang process == aemp port), so many of the documents and
697programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 776programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
698sample: 777sample:
699 778
700 http://www.Erlang.se/doc/programming_rules.shtml 779 http://www.erlang.se/doc/programming_rules.shtml
701 http://Erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4 780 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
702 http://Erlang.org/download/Erlang-book-part1.pdf # chapters 5 and 6 781 http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6
703 http://Erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5 782 http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
704 783
705Despite the similarities, there are also some important differences: 784Despite the similarities, there are also some important differences:
706 785
707=over 4 786=over 4
708 787
709=item * Node IDs are arbitrary strings in AEMP. 788=item * Node IDs are arbitrary strings in AEMP.
710 789
711Erlang relies on special naming and DNS to work everywhere in the same 790Erlang relies on special naming and DNS to work everywhere in the same
712way. AEMP relies on each node somehow knowing its own address(es) (e.g. by 791way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
713configuraiton or DNS), but will otherwise discover other odes itself. 792configuration or DNS), but will otherwise discover other odes itself.
714 793
715=item * Erlang has a "remote ports are like local ports" philosophy, AEMP 794=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
716uses "local ports are like remote ports". 795uses "local ports are like remote ports".
717 796
718The failure modes for local ports are quite different (runtime errors 797The failure modes for local ports are quite different (runtime errors
731 810
732Erlang uses processes that selectively receive messages, and therefore 811Erlang uses processes that selectively receive messages, and therefore
733needs a queue. AEMP is event based, queuing messages would serve no 812needs a queue. AEMP is event based, queuing messages would serve no
734useful purpose. For the same reason the pattern-matching abilities of 813useful purpose. For the same reason the pattern-matching abilities of
735AnyEvent::MP are more limited, as there is little need to be able to 814AnyEvent::MP are more limited, as there is little need to be able to
736filter messages without dequeing them. 815filter messages without dequeuing them.
737 816
738(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 817(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
739 818
740=item * Erlang sends are synchronous, AEMP sends are asynchronous. 819=item * Erlang sends are synchronous, AEMP sends are asynchronous.
741 820
847L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 926L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
848 927
849L<AnyEvent::MP::Global> - network maintainance and port groups, to find 928L<AnyEvent::MP::Global> - network maintainance and port groups, to find
850your applications. 929your applications.
851 930
931L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
932all nodes.
933
852L<AnyEvent>. 934L<AnyEvent>.
853 935
854=head1 AUTHOR 936=head1 AUTHOR
855 937
856 Marc Lehmann <schmorp@schmorp.de> 938 Marc Lehmann <schmorp@schmorp.de>

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines