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.33 by root, Wed Aug 5 22:40:51 2009 UTC vs.
Revision 1.43 by root, Sun Aug 9 16:08:16 2009 UTC

8 8
9 $NODE # contains this node's noderef 9 $NODE # contains this node's noderef
10 NODE # returns this node's noderef 10 NODE # returns this node's noderef
11 NODE $port # returns the noderef of the port 11 NODE $port # returns the noderef of the port
12 12
13 $SELF # receiving/own port id in rcv callbacks
14
15 # ports are message endpoints
16
17 # sending messages
13 snd $port, type => data...; 18 snd $port, type => data...;
19 snd $port, @msg;
20 snd @msg_with_first_element_being_a_port;
14 21
15 $SELF # receiving/own port id in rcv callbacks 22 # miniports
23 my $miniport = port { my @msg = @_; 0 };
16 24
25 # full ports
26 my $port = port;
17 rcv $port, smartmatch => $cb->($port, @msg); 27 rcv $port, smartmatch => $cb->(@msg);
18
19 # examples:
20 rcv $port2, ping => sub { snd $_[0], "pong"; 0 }; 28 rcv $port, ping => sub { snd $_[0], "pong"; 0 };
21 rcv $port1, pong => sub { warn "pong received\n" }; 29 rcv $port, pong => sub { warn "pong received\n"; 0 };
22 snd $port2, ping => $port1; 30
31 # remote ports
32 my $port = spawn $node, $initfunc, @initdata;
23 33
24 # more, smarter, matches (_any_ is exported by this module) 34 # more, smarter, matches (_any_ is exported by this module)
25 rcv $port, [child_died => $pid] => sub { ... 35 rcv $port, [child_died => $pid] => sub { ...
26 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3 36 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3
37
38 # monitoring
39 mon $port, $cb->(@msg) # callback is invoked on death
40 mon $port, $otherport # kill otherport on abnormal death
41 mon $port, $otherport, @msg # send message on death
27 42
28=head1 DESCRIPTION 43=head1 DESCRIPTION
29 44
30This module (-family) implements a simple message passing framework. 45This module (-family) implements a simple message passing framework.
31 46
100 115
101use AE (); 116use AE ();
102 117
103use base "Exporter"; 118use base "Exporter";
104 119
105our $VERSION = '0.1'; 120our $VERSION = $AnyEvent::MP::Base::VERSION;
121
106our @EXPORT = qw( 122our @EXPORT = qw(
107 NODE $NODE *SELF node_of _any_ 123 NODE $NODE *SELF node_of _any_
108 resolve_node initialise_node 124 resolve_node initialise_node
109 snd rcv mon kil reg psub 125 snd rcv mon kil reg psub spawn
110 port 126 port
111); 127);
112 128
113our $SELF; 129our $SELF;
114 130
127 143
128=item $noderef = node_of $port 144=item $noderef = node_of $port
129 145
130Extracts and returns the noderef from a portid or a noderef. 146Extracts and returns the noderef from a portid or a noderef.
131 147
148=item initialise_node $noderef, $seednode, $seednode...
149
150=item initialise_node "slave/", $master, $master...
151
152Before a node can talk to other nodes on the network it has to initialise
153itself - the minimum a node needs to know is it's own name, and optionally
154it should know the noderefs of some other nodes in the network.
155
156This function initialises a node - it must be called exactly once (or
157never) before calling other AnyEvent::MP functions.
158
159All arguments are noderefs, which can be either resolved or unresolved.
160
161There are two types of networked nodes, public nodes and slave nodes:
162
163=over 4
164
165=item public nodes
166
167For public nodes, C<$noderef> must either be a (possibly unresolved)
168noderef, in which case it will be resolved, or C<undef> (or missing), in
169which case the noderef will be guessed.
170
171Afterwards, the node will bind itself on all endpoints and try to connect
172to all additional C<$seednodes> that are specified. Seednodes are optional
173and can be used to quickly bootstrap the node into an existing network.
174
175=item slave nodes
176
177When the C<$noderef> is the special string C<slave/>, then the node will
178become a slave node. Slave nodes cannot be contacted from outside and will
179route most of their traffic to the master node that they attach to.
180
181At least one additional noderef is required: The node will try to connect
182to all of them and will become a slave attached to the first node it can
183successfully connect to.
184
185=back
186
187This function will block until all nodes have been resolved and, for slave
188nodes, until it has successfully established a connection to a master
189server.
190
191Example: become a public node listening on the default node.
192
193 initialise_node;
194
195Example: become a public node, and try to contact some well-known master
196servers to become part of the network.
197
198 initialise_node undef, "master1", "master2";
199
200Example: become a public node listening on port C<4041>.
201
202 initialise_node 4041;
203
204Example: become a public node, only visible on localhost port 4044.
205
206 initialise_node "locahost:4044";
207
208Example: become a slave node to any of the specified master servers.
209
210 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
211
132=item $cv = resolve_node $noderef 212=item $cv = resolve_node $noderef
133 213
134Takes an unresolved node reference that may contain hostnames and 214Takes an unresolved node reference that may contain hostnames and
135abbreviated IDs, resolves all of them and returns a resolved node 215abbreviated IDs, resolves all of them and returns a resolved node
136reference. 216reference.
233 $port 313 $port
234} 314}
235 315
236=item reg $port, $name 316=item reg $port, $name
237 317
238Registers the given port under the name C<$name>. If the name already 318=item reg $name
239exists it is replaced. 319
320Registers the given port (or C<$SELF><<< if missing) under the name
321C<$name>. If the name already exists it is replaced.
240 322
241A port can only be registered under one well known name. 323A port can only be registered under one well known name.
242 324
243A port automatically becomes unregistered when it is killed. 325A port automatically becomes unregistered when it is killed.
244 326
245=cut 327=cut
246 328
247sub reg(@) { 329sub reg(@) {
248 my ($port, $name) = @_; 330 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
249 331
250 $REG{$name} = $port; 332 $REG{$_[0]} = $port;
251} 333}
252 334
253=item rcv $port, $callback->(@msg) 335=item rcv $port, $callback->(@msg)
254 336
255Replaces the callback on the specified miniport (after converting it to 337Replaces the callback on the specified miniport (after converting it to
260=item rcv $port, $smartmatch => $callback->(@msg), ... 342=item rcv $port, $smartmatch => $callback->(@msg), ...
261 343
262=item rcv $port, [$smartmatch...] => $callback->(@msg), ... 344=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
263 345
264Register callbacks to be called on matching messages on the given full 346Register callbacks to be called on matching messages on the given full
265port (after converting it to one if required). 347port (after converting it to one if required) and return the port.
266 348
267The callback has to return a true value when its work is done, after 349The callback has to return a true value when its work is done, after
268which is will be removed, or a false value in which case it will stay 350which is will be removed, or a false value in which case it will stay
269registered. 351registered.
270 352
271The global C<$SELF> (exported by this module) contains C<$port> while 353The global C<$SELF> (exported by this module) contains C<$port> while
272executing the callback. 354executing the callback.
273 355
274Runtime errors wdurign callback execution will result in the port being 356Runtime errors during callback execution will result in the port being
275C<kil>ed. 357C<kil>ed.
276 358
277If the match is an array reference, then it will be matched against the 359If the match is an array reference, then it will be matched against the
278first elements of the message, otherwise only the first element is being 360first elements of the message, otherwise only the first element is being
279matched. 361matched.
282exported by this module) matches any single element of the message. 364exported by this module) matches any single element of the message.
283 365
284While not required, it is highly recommended that the first matching 366While not required, it is highly recommended that the first matching
285element is a string identifying the message. The one-string-only match is 367element is a string identifying the message. The one-string-only match is
286also the most efficient match (by far). 368also the most efficient match (by far).
369
370Example: create a port and bind receivers on it in one go.
371
372 my $port = rcv port,
373 msg1 => sub { ...; 0 },
374 msg2 => sub { ...; 0 },
375 ;
376
377Example: create a port, bind receivers and send it in a message elsewhere
378in one go:
379
380 snd $otherport, reply =>
381 rcv port,
382 msg1 => sub { ...; 0 },
383 ...
384 ;
287 385
288=cut 386=cut
289 387
290sub rcv($@) { 388sub rcv($@) {
291 my $port = shift; 389 my $port = shift;
398 } 496 }
399} 497}
400 498
401=item $guard = mon $port, $cb->(@reason) 499=item $guard = mon $port, $cb->(@reason)
402 500
403=item $guard = mon $port, $otherport 501=item $guard = mon $port, $rcvport
404 502
503=item $guard = mon $port
504
405=item $guard = mon $port, $otherport, @msg 505=item $guard = mon $port, $rcvport, @msg
406 506
407Monitor the given port and do something when the port is killed. 507Monitor the given port and do something when the port is killed or
508messages to it were lost, and optionally return a guard that can be used
509to stop monitoring again.
408 510
511C<mon> effectively guarantees that, in the absence of hardware failures,
512that after starting the monitor, either all messages sent to the port
513will arrive, or the monitoring action will be invoked after possible
514message loss has been detected. No messages will be lost "in between"
515(after the first lost message no further messages will be received by the
516port). After the monitoring action was invoked, further messages might get
517delivered again.
518
409In the first form, the callback is simply called with any number 519In the first form (callback), the callback is simply called with any
410of C<@reason> elements (no @reason means that the port was deleted 520number of C<@reason> elements (no @reason means that the port was deleted
411"normally"). Note also that I<< the callback B<must> never die >>, so use 521"normally"). Note also that I<< the callback B<must> never die >>, so use
412C<eval> if unsure. 522C<eval> if unsure.
413 523
414In the second form, the other port will be C<kil>'ed with C<@reason>, iff 524In the second form (another port given), the other port (C<$rcvport>)
415a @reason was specified, i.e. on "normal" kils nothing happens, while 525will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on
416under all other conditions, the other port is killed with the same reason. 526"normal" kils nothing happens, while under all other conditions, the other
527port is killed with the same reason.
417 528
529The third form (kill self) is the same as the second form, except that
530C<$rvport> defaults to C<$SELF>.
531
418In the last form, a message of the form C<@msg, @reason> will be C<snd>. 532In the last form (message), a message of the form C<@msg, @reason> will be
533C<snd>.
534
535As a rule of thumb, monitoring requests should always monitor a port from
536a local port (or callback). The reason is that kill messages might get
537lost, just like any other message. Another less obvious reason is that
538even monitoring requests can get lost (for exmaple, when the connection
539to the other node goes down permanently). When monitoring a port locally
540these problems do not exist.
419 541
420Example: call a given callback when C<$port> is killed. 542Example: call a given callback when C<$port> is killed.
421 543
422 mon $port, sub { warn "port died because of <@_>\n" }; 544 mon $port, sub { warn "port died because of <@_>\n" };
423 545
424Example: kill ourselves when C<$port> is killed abnormally. 546Example: kill ourselves when C<$port> is killed abnormally.
425 547
426 mon $port, $self; 548 mon $port;
427 549
428Example: send us a restart message another C<$port> is killed. 550Example: send us a restart message when another C<$port> is killed.
429 551
430 mon $port, $self => "restart"; 552 mon $port, $self => "restart";
431 553
432=cut 554=cut
433 555
434sub mon { 556sub mon {
435 my ($noderef, $port) = split /#/, shift, 2; 557 my ($noderef, $port) = split /#/, shift, 2;
436 558
437 my $node = $NODE{$noderef} || add_node $noderef; 559 my $node = $NODE{$noderef} || add_node $noderef;
438 560
439 my $cb = shift; 561 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
440 562
441 unless (ref $cb) { 563 unless (ref $cb) {
442 if (@_) { 564 if (@_) {
443 # send a kill info message 565 # send a kill info message
444 my (@msg) = ($cb, @_); 566 my (@msg) = ($cb, @_);
475=cut 597=cut
476 598
477sub mon_guard { 599sub mon_guard {
478 my ($port, @refs) = @_; 600 my ($port, @refs) = @_;
479 601
602 #TODO: mon-less form?
603
480 mon $port, sub { 0 && @refs } 604 mon $port, sub { 0 && @refs }
481} 605}
482 606
483=item lnk $port1, $port2
484
485Link two ports. This is simply a shorthand for:
486
487 mon $port1, $port2;
488 mon $port2, $port1;
489
490It means that if either one is killed abnormally, the other one gets
491killed as well.
492
493=item kil $port[, @reason] 607=item kil $port[, @reason]
494 608
495Kill the specified port with the given C<@reason>. 609Kill the specified port with the given C<@reason>.
496 610
497If no C<@reason> is specified, then the port is killed "normally" (linked 611If no C<@reason> is specified, then the port is killed "normally" (linked
504will be reported as reason C<< die => $@ >>. 618will be reported as reason C<< die => $@ >>.
505 619
506Transport/communication errors are reported as C<< transport_error => 620Transport/communication errors are reported as C<< transport_error =>
507$message >>. 621$message >>.
508 622
509=back
510
511=head1 FUNCTIONS FOR NODES
512
513=over 4
514
515=item initialise_node $noderef, $seednode, $seednode...
516
517=item initialise_node "slave/", $master, $master...
518
519Initialises a node - must be called exactly once before calling other
520AnyEvent::MP functions when talking to other nodes is required.
521
522All arguments are noderefs, which can be either resolved or unresolved.
523
524There are two types of networked nodes, public nodes and slave nodes:
525
526=over 4
527
528=item public nodes
529
530For public nodes, C<$noderef> must either be a (possibly unresolved)
531noderef, in which case it will be resolved, or C<undef> (or missing), in
532which case the noderef will be guessed.
533
534Afterwards, the node will bind itself on all endpoints and try to connect
535to all additional C<$seednodes> that are specified. Seednodes are optional
536and can be used to quickly bootstrap the node into an existing network.
537
538=item slave nodes
539
540When the C<$noderef> is the special string C<slave/>, then the node will
541become a slave node. Slave nodes cannot be contacted from outside and will
542route most of their traffic to the master node that they attach to.
543
544At least one additional noderef is required: The node will try to connect
545to all of them and will become a slave attached to the first node it can
546successfully connect to.
547
548=back
549
550This function will block until all nodes have been resolved and, for slave
551nodes, until it has successfully established a connection to a master
552server.
553
554Example: become a public node listening on the default node.
555
556 initialise_node;
557
558Example: become a public node, and try to contact some well-known master
559servers to become part of the network.
560
561 initialise_node undef, "master1", "master2";
562
563Example: become a public node listening on port C<4041>.
564
565 initialise_node 4041;
566
567Example: become a public node, only visible on localhost port 4044.
568
569 initialise_node "locahost:4044";
570
571Example: become a slave node to any of the specified master servers.
572
573 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
574
575=cut 623=cut
624
625=item $port = spawn $node, $initfunc[, @initdata]
626
627Creates a port on the node C<$node> (which can also be a port ID, in which
628case it's the node where that port resides).
629
630The port ID of the newly created port is return immediately, and it is
631permissible to immediately start sending messages or monitor the port.
632
633After the port has been created, the init function is
634called. This function must be a fully-qualified function name
635(e.g. C<MyApp::Chat::Server::init>). To specify a function in the main
636program, use C<::name>.
637
638If the function doesn't exist, then the node tries to C<require>
639the package, then the package above the package and so on (e.g.
640C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
641exists or it runs out of package names.
642
643The init function is then called with the newly-created port as context
644object (C<$SELF>) and the C<@initdata> values as arguments.
645
646A common idiom is to pass your own port, monitor the spawned port, and
647in the init function, monitor the original port. This two-way monitoring
648ensures that both ports get cleaned up when there is a problem.
649
650Example: spawn a chat server port on C<$othernode>.
651
652 # this node, executed from within a port context:
653 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
654 mon $server;
655
656 # init function on C<$othernode>
657 sub connect {
658 my ($srcport) = @_;
659
660 mon $srcport;
661
662 rcv $SELF, sub {
663 ...
664 };
665 }
666
667=cut
668
669sub _spawn {
670 my $port = shift;
671 my $init = shift;
672
673 local $SELF = "$NODE#$port";
674 eval {
675 &{ load_func $init }
676 };
677 _self_die if $@;
678}
679
680sub spawn(@) {
681 my ($noderef, undef) = split /#/, shift, 2;
682
683 my $id = "$RUNIQ." . $ID++;
684
685 $_[0] =~ /::/
686 or Carp::croak "spawn init function must be a fully-qualified name, caught";
687
688 ($NODE{$noderef} || add_node $noderef)
689 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
690
691 "$noderef#$id"
692}
576 693
577=back 694=back
578 695
579=head1 NODE MESSAGES 696=head1 NODE MESSAGES
580 697
622 739
623=back 740=back
624 741
625=head1 AnyEvent::MP vs. Distributed Erlang 742=head1 AnyEvent::MP vs. Distributed Erlang
626 743
627AnyEvent::MP got lots of its ideas from distributed erlang (erlang node 744AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
628== aemp node, erlang process == aemp port), so many of the documents and 745== aemp node, Erlang process == aemp port), so many of the documents and
629programming techniques employed by erlang apply to AnyEvent::MP. Here is a 746programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
630sample: 747sample:
631 748
632 http://www.erlang.se/doc/programming_rules.shtml 749 http://www.Erlang.se/doc/programming_rules.shtml
633 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4 750 http://Erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
634 http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6 751 http://Erlang.org/download/Erlang-book-part1.pdf # chapters 5 and 6
635 http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5 752 http://Erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
636 753
637Despite the similarities, there are also some important differences: 754Despite the similarities, there are also some important differences:
638 755
639=over 4 756=over 4
640 757
651 768
652Erlang uses processes that selctively receive messages, and therefore 769Erlang uses processes that selctively receive messages, and therefore
653needs a queue. AEMP is event based, queuing messages would serve no useful 770needs a queue. AEMP is event based, queuing messages would serve no useful
654purpose. 771purpose.
655 772
656(But see L<Coro::MP> for a more erlang-like process model on top of AEMP). 773(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
657 774
658=item * Erlang sends are synchronous, AEMP sends are asynchronous. 775=item * Erlang sends are synchronous, AEMP sends are asynchronous.
659 776
660Sending messages in erlang is synchronous and blocks the process. AEMP 777Sending messages in Erlang is synchronous and blocks the process. AEMP
661sends are immediate, connection establishment is handled in the 778sends are immediate, connection establishment is handled in the
662background. 779background.
663 780
664=item * Erlang can silently lose messages, AEMP cannot. 781=item * Erlang can silently lose messages, AEMP cannot.
665 782
668and c, and the other side only receives messages a and c). 785and c, and the other side only receives messages a and c).
669 786
670AEMP guarantees correct ordering, and the guarantee that there are no 787AEMP guarantees correct ordering, and the guarantee that there are no
671holes in the message sequence. 788holes in the message sequence.
672 789
673=item * In erlang, processes can be declared dead and later be found to be 790=item * In Erlang, processes can be declared dead and later be found to be
674alive. 791alive.
675 792
676In erlang it can happen that a monitored process is declared dead and 793In Erlang it can happen that a monitored process is declared dead and
677linked processes get killed, but later it turns out that the process is 794linked processes get killed, but later it turns out that the process is
678still alive - and can receive messages. 795still alive - and can receive messages.
679 796
680In AEMP, when port monitoring detects a port as dead, then that port will 797In AEMP, when port monitoring detects a port as dead, then that port will
681eventually be killed - it cannot happen that a node detects a port as dead 798eventually be killed - it cannot happen that a node detects a port as dead
682and then later sends messages to it, finding it is still alive. 799and then later sends messages to it, finding it is still alive.
683 800
684=item * Erlang can send messages to the wrong port, AEMP does not. 801=item * Erlang can send messages to the wrong port, AEMP does not.
685 802
686In erlang it is quite possible that a node that restarts reuses a process 803In Erlang it is quite possible that a node that restarts reuses a process
687ID known to other nodes for a completely different process, causing 804ID known to other nodes for a completely different process, causing
688messages destined for that process to end up in an unrelated process. 805messages destined for that process to end up in an unrelated process.
689 806
690AEMP never reuses port IDs, so old messages or old port IDs floating 807AEMP never reuses port IDs, so old messages or old port IDs floating
691around in the network will not be sent to an unrelated port. 808around in the network will not be sent to an unrelated port.
697securely authenticate nodes. 814securely authenticate nodes.
698 815
699=item * The AEMP protocol is optimised for both text-based and binary 816=item * The AEMP protocol is optimised for both text-based and binary
700communications. 817communications.
701 818
702The AEMP protocol, unlike the erlang protocol, supports both 819The AEMP protocol, unlike the Erlang protocol, supports both
703language-independent text-only protocols (good for debugging) and binary, 820language-independent text-only protocols (good for debugging) and binary,
704language-specific serialisers (e.g. Storable). 821language-specific serialisers (e.g. Storable).
705 822
706It has also been carefully designed to be implementable in other languages 823It has also been carefully designed to be implementable in other languages
707with a minimum of work while gracefully degrading fucntionality to make the 824with a minimum of work while gracefully degrading fucntionality to make the
708protocol simple. 825protocol simple.
709 826
827=item * AEMP has more flexible monitoring options than Erlang.
828
829In Erlang, you can chose to receive I<all> exit signals as messages
830or I<none>, there is no in-between, so monitoring single processes is
831difficult to implement. Monitoring in AEMP is more flexible than in
832Erlang, as one can choose between automatic kill, exit message or callback
833on a per-process basis.
834
835=item * Erlang tries to hide remote/local connections, AEMP does not.
836
837Monitoring in Erlang is not an indicator of process death/crashes,
838as linking is (except linking is unreliable in Erlang).
839
840In AEMP, you don't "look up" registered port names or send to named ports
841that might or might not be persistent. Instead, you normally spawn a port
842on the remote node. The init function monitors the you, and you monitor
843the remote port. Since both monitors are local to the node, they are much
844more reliable.
845
846This also saves round-trips and avoids sending messages to the wrong port
847(hard to do in Erlang).
848
710=back 849=back
711 850
712=head1 SEE ALSO 851=head1 SEE ALSO
713 852
714L<AnyEvent>. 853L<AnyEvent>.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines