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

Comparing cvsroot/AnyEvent-MP/MP.pm (file contents):
Revision 1.34 by root, Wed Aug 5 23:50:46 2009 UTC vs.
Revision 1.42 by root, Sun Aug 9 00:41:49 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
104 119
105our $VERSION = '0.1'; 120our $VERSION = '0.1';
106our @EXPORT = qw( 121our @EXPORT = qw(
107 NODE $NODE *SELF node_of _any_ 122 NODE $NODE *SELF node_of _any_
108 resolve_node initialise_node 123 resolve_node initialise_node
109 snd rcv mon kil reg psub 124 snd rcv mon kil reg psub spawn
110 port 125 port
111); 126);
112 127
113our $SELF; 128our $SELF;
114 129
297 $port 312 $port
298} 313}
299 314
300=item reg $port, $name 315=item reg $port, $name
301 316
302Registers the given port under the name C<$name>. If the name already 317=item reg $name
303exists it is replaced. 318
319Registers the given port (or C<$SELF><<< if missing) under the name
320C<$name>. If the name already exists it is replaced.
304 321
305A port can only be registered under one well known name. 322A port can only be registered under one well known name.
306 323
307A port automatically becomes unregistered when it is killed. 324A port automatically becomes unregistered when it is killed.
308 325
309=cut 326=cut
310 327
311sub reg(@) { 328sub reg(@) {
312 my ($port, $name) = @_; 329 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
313 330
314 $REG{$name} = $port; 331 $REG{$_[0]} = $port;
315} 332}
316 333
317=item rcv $port, $callback->(@msg) 334=item rcv $port, $callback->(@msg)
318 335
319Replaces the callback on the specified miniport (after converting it to 336Replaces the callback on the specified miniport (after converting it to
324=item rcv $port, $smartmatch => $callback->(@msg), ... 341=item rcv $port, $smartmatch => $callback->(@msg), ...
325 342
326=item rcv $port, [$smartmatch...] => $callback->(@msg), ... 343=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
327 344
328Register callbacks to be called on matching messages on the given full 345Register callbacks to be called on matching messages on the given full
329port (after converting it to one if required). 346port (after converting it to one if required) and return the port.
330 347
331The callback has to return a true value when its work is done, after 348The callback has to return a true value when its work is done, after
332which is will be removed, or a false value in which case it will stay 349which is will be removed, or a false value in which case it will stay
333registered. 350registered.
334 351
335The global C<$SELF> (exported by this module) contains C<$port> while 352The global C<$SELF> (exported by this module) contains C<$port> while
336executing the callback. 353executing the callback.
337 354
338Runtime errors wdurign callback execution will result in the port being 355Runtime errors during callback execution will result in the port being
339C<kil>ed. 356C<kil>ed.
340 357
341If the match is an array reference, then it will be matched against the 358If the match is an array reference, then it will be matched against the
342first elements of the message, otherwise only the first element is being 359first elements of the message, otherwise only the first element is being
343matched. 360matched.
346exported by this module) matches any single element of the message. 363exported by this module) matches any single element of the message.
347 364
348While not required, it is highly recommended that the first matching 365While not required, it is highly recommended that the first matching
349element is a string identifying the message. The one-string-only match is 366element is a string identifying the message. The one-string-only match is
350also the most efficient match (by far). 367also the most efficient match (by far).
368
369Example: create a port and bind receivers on it in one go.
370
371 my $port = rcv port,
372 msg1 => sub { ...; 0 },
373 msg2 => sub { ...; 0 },
374 ;
375
376Example: create a port, bind receivers and send it in a message elsewhere
377in one go:
378
379 snd $otherport, reply =>
380 rcv port,
381 msg1 => sub { ...; 0 },
382 ...
383 ;
351 384
352=cut 385=cut
353 386
354sub rcv($@) { 387sub rcv($@) {
355 my $port = shift; 388 my $port = shift;
462 } 495 }
463} 496}
464 497
465=item $guard = mon $port, $cb->(@reason) 498=item $guard = mon $port, $cb->(@reason)
466 499
467=item $guard = mon $port, $otherport 500=item $guard = mon $port, $rcvport
468 501
502=item $guard = mon $port
503
469=item $guard = mon $port, $otherport, @msg 504=item $guard = mon $port, $rcvport, @msg
470 505
471Monitor the given port and do something when the port is killed. 506Monitor the given port and do something when the port is killed or
507messages to it were lost, and optionally return a guard that can be used
508to stop monitoring again.
472 509
510C<mon> effectively guarantees that, in the absence of hardware failures,
511that after starting the monitor, either all messages sent to the port
512will arrive, or the monitoring action will be invoked after possible
513message loss has been detected. No messages will be lost "in between"
514(after the first lost message no further messages will be received by the
515port). After the monitoring action was invoked, further messages might get
516delivered again.
517
473In the first form, the callback is simply called with any number 518In the first form (callback), the callback is simply called with any
474of C<@reason> elements (no @reason means that the port was deleted 519number of C<@reason> elements (no @reason means that the port was deleted
475"normally"). Note also that I<< the callback B<must> never die >>, so use 520"normally"). Note also that I<< the callback B<must> never die >>, so use
476C<eval> if unsure. 521C<eval> if unsure.
477 522
478In the second form, the other port will be C<kil>'ed with C<@reason>, iff 523In the second form (another port given), the other port (C<$rcvport)
479a @reason was specified, i.e. on "normal" kils nothing happens, while 524will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on
480under all other conditions, the other port is killed with the same reason. 525"normal" kils nothing happens, while under all other conditions, the other
526port is killed with the same reason.
481 527
528The third form (kill self) is the same as the second form, except that
529C<$rvport> defaults to C<$SELF>.
530
482In the last form, a message of the form C<@msg, @reason> will be C<snd>. 531In the last form (message), a message of the form C<@msg, @reason> will be
532C<snd>.
533
534As a rule of thumb, monitoring requests should always monitor a port from
535a local port (or callback). The reason is that kill messages might get
536lost, just like any other message. Another less obvious reason is that
537even monitoring requests can get lost (for exmaple, when the connection
538to the other node goes down permanently). When monitoring a port locally
539these problems do not exist.
483 540
484Example: call a given callback when C<$port> is killed. 541Example: call a given callback when C<$port> is killed.
485 542
486 mon $port, sub { warn "port died because of <@_>\n" }; 543 mon $port, sub { warn "port died because of <@_>\n" };
487 544
488Example: kill ourselves when C<$port> is killed abnormally. 545Example: kill ourselves when C<$port> is killed abnormally.
489 546
490 mon $port, $self; 547 mon $port;
491 548
492Example: send us a restart message another C<$port> is killed. 549Example: send us a restart message when another C<$port> is killed.
493 550
494 mon $port, $self => "restart"; 551 mon $port, $self => "restart";
495 552
496=cut 553=cut
497 554
498sub mon { 555sub mon {
499 my ($noderef, $port) = split /#/, shift, 2; 556 my ($noderef, $port) = split /#/, shift, 2;
500 557
501 my $node = $NODE{$noderef} || add_node $noderef; 558 my $node = $NODE{$noderef} || add_node $noderef;
502 559
503 my $cb = shift; 560 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
504 561
505 unless (ref $cb) { 562 unless (ref $cb) {
506 if (@_) { 563 if (@_) {
507 # send a kill info message 564 # send a kill info message
508 my (@msg) = ($cb, @_); 565 my (@msg) = ($cb, @_);
539=cut 596=cut
540 597
541sub mon_guard { 598sub mon_guard {
542 my ($port, @refs) = @_; 599 my ($port, @refs) = @_;
543 600
601 #TODO: mon-less form?
602
544 mon $port, sub { 0 && @refs } 603 mon $port, sub { 0 && @refs }
545} 604}
546 605
547=item lnk $port1, $port2
548
549Link two ports. This is simply a shorthand for:
550
551 mon $port1, $port2;
552 mon $port2, $port1;
553
554It means that if either one is killed abnormally, the other one gets
555killed as well.
556
557=item kil $port[, @reason] 606=item kil $port[, @reason]
558 607
559Kill the specified port with the given C<@reason>. 608Kill the specified port with the given C<@reason>.
560 609
561If no C<@reason> is specified, then the port is killed "normally" (linked 610If no C<@reason> is specified, then the port is killed "normally" (linked
567Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 616Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
568will be reported as reason C<< die => $@ >>. 617will be reported as reason C<< die => $@ >>.
569 618
570Transport/communication errors are reported as C<< transport_error => 619Transport/communication errors are reported as C<< transport_error =>
571$message >>. 620$message >>.
621
622=cut
623
624=item $port = spawn $node, $initfunc[, @initdata]
625
626Creates a port on the node C<$node> (which can also be a port ID, in which
627case it's the node where that port resides).
628
629The port ID of the newly created port is return immediately, and it is
630permissible to immediately start sending messages or monitor the port.
631
632After the port has been created, the init function is
633called. This function must be a fully-qualified function name
634(e.g. C<MyApp::Chat::Server::init>). To specify a function in the main
635program, use C<::name>.
636
637If the function doesn't exist, then the node tries to C<require>
638the package, then the package above the package and so on (e.g.
639C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
640exists or it runs out of package names.
641
642The init function is then called with the newly-created port as context
643object (C<$SELF>) and the C<@initdata> values as arguments.
644
645A common idiom is to pass your own port, monitor the spawned port, and
646in the init function, monitor the original port. This two-way monitoring
647ensures that both ports get cleaned up when there is a problem.
648
649Example: spawn a chat server port on C<$othernode>.
650
651 # this node, executed from within a port context:
652 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
653 mon $server;
654
655 # init function on C<$othernode>
656 sub connect {
657 my ($srcport) = @_;
658
659 mon $srcport;
660
661 rcv $SELF, sub {
662 ...
663 };
664 }
665
666=cut
667
668sub _spawn {
669 my $port = shift;
670 my $init = shift;
671
672 local $SELF = "$NODE#$port";
673 eval {
674 &{ load_func $init }
675 };
676 _self_die if $@;
677}
678
679sub spawn(@) {
680 my ($noderef, undef) = split /#/, shift, 2;
681
682 my $id = "$RUNIQ." . $ID++;
683
684 $_[0] =~ /::/
685 or Carp::croak "spawn init function must be a fully-qualified name, caught";
686
687 ($NODE{$noderef} || add_node $noderef)
688 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
689
690 "$noderef#$id"
691}
572 692
573=back 693=back
574 694
575=head1 NODE MESSAGES 695=head1 NODE MESSAGES
576 696
618 738
619=back 739=back
620 740
621=head1 AnyEvent::MP vs. Distributed Erlang 741=head1 AnyEvent::MP vs. Distributed Erlang
622 742
623AnyEvent::MP got lots of its ideas from distributed erlang (erlang node 743AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
624== aemp node, erlang process == aemp port), so many of the documents and 744== aemp node, Erlang process == aemp port), so many of the documents and
625programming techniques employed by erlang apply to AnyEvent::MP. Here is a 745programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
626sample: 746sample:
627 747
628 http://www.erlang.se/doc/programming_rules.shtml 748 http://www.Erlang.se/doc/programming_rules.shtml
629 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4 749 http://Erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
630 http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6 750 http://Erlang.org/download/Erlang-book-part1.pdf # chapters 5 and 6
631 http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5 751 http://Erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
632 752
633Despite the similarities, there are also some important differences: 753Despite the similarities, there are also some important differences:
634 754
635=over 4 755=over 4
636 756
647 767
648Erlang uses processes that selctively receive messages, and therefore 768Erlang uses processes that selctively receive messages, and therefore
649needs a queue. AEMP is event based, queuing messages would serve no useful 769needs a queue. AEMP is event based, queuing messages would serve no useful
650purpose. 770purpose.
651 771
652(But see L<Coro::MP> for a more erlang-like process model on top of AEMP). 772(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
653 773
654=item * Erlang sends are synchronous, AEMP sends are asynchronous. 774=item * Erlang sends are synchronous, AEMP sends are asynchronous.
655 775
656Sending messages in erlang is synchronous and blocks the process. AEMP 776Sending messages in Erlang is synchronous and blocks the process. AEMP
657sends are immediate, connection establishment is handled in the 777sends are immediate, connection establishment is handled in the
658background. 778background.
659 779
660=item * Erlang can silently lose messages, AEMP cannot. 780=item * Erlang can silently lose messages, AEMP cannot.
661 781
664and c, and the other side only receives messages a and c). 784and c, and the other side only receives messages a and c).
665 785
666AEMP guarantees correct ordering, and the guarantee that there are no 786AEMP guarantees correct ordering, and the guarantee that there are no
667holes in the message sequence. 787holes in the message sequence.
668 788
669=item * In erlang, processes can be declared dead and later be found to be 789=item * In Erlang, processes can be declared dead and later be found to be
670alive. 790alive.
671 791
672In erlang it can happen that a monitored process is declared dead and 792In Erlang it can happen that a monitored process is declared dead and
673linked processes get killed, but later it turns out that the process is 793linked processes get killed, but later it turns out that the process is
674still alive - and can receive messages. 794still alive - and can receive messages.
675 795
676In AEMP, when port monitoring detects a port as dead, then that port will 796In AEMP, when port monitoring detects a port as dead, then that port will
677eventually be killed - it cannot happen that a node detects a port as dead 797eventually be killed - it cannot happen that a node detects a port as dead
678and then later sends messages to it, finding it is still alive. 798and then later sends messages to it, finding it is still alive.
679 799
680=item * Erlang can send messages to the wrong port, AEMP does not. 800=item * Erlang can send messages to the wrong port, AEMP does not.
681 801
682In erlang it is quite possible that a node that restarts reuses a process 802In Erlang it is quite possible that a node that restarts reuses a process
683ID known to other nodes for a completely different process, causing 803ID known to other nodes for a completely different process, causing
684messages destined for that process to end up in an unrelated process. 804messages destined for that process to end up in an unrelated process.
685 805
686AEMP never reuses port IDs, so old messages or old port IDs floating 806AEMP never reuses port IDs, so old messages or old port IDs floating
687around in the network will not be sent to an unrelated port. 807around in the network will not be sent to an unrelated port.
693securely authenticate nodes. 813securely authenticate nodes.
694 814
695=item * The AEMP protocol is optimised for both text-based and binary 815=item * The AEMP protocol is optimised for both text-based and binary
696communications. 816communications.
697 817
698The AEMP protocol, unlike the erlang protocol, supports both 818The AEMP protocol, unlike the Erlang protocol, supports both
699language-independent text-only protocols (good for debugging) and binary, 819language-independent text-only protocols (good for debugging) and binary,
700language-specific serialisers (e.g. Storable). 820language-specific serialisers (e.g. Storable).
701 821
702It has also been carefully designed to be implementable in other languages 822It has also been carefully designed to be implementable in other languages
703with a minimum of work while gracefully degrading fucntionality to make the 823with a minimum of work while gracefully degrading fucntionality to make the
704protocol simple. 824protocol simple.
705 825
826=item * AEMP has more flexible monitoring options than Erlang.
827
828In Erlang, you can chose to receive I<all> exit signals as messages
829or I<none>, there is no in-between, so monitoring single processes is
830difficult to implement. Monitoring in AEMP is more flexible than in
831Erlang, as one can choose between automatic kill, exit message or callback
832on a per-process basis.
833
834=item * Erlang tries to hide remote/local connections, AEMP does not.
835
836Monitoring in Erlang is not an indicator of process death/crashes,
837as linking is (except linking is unreliable in Erlang).
838
839In AEMP, you don't "look up" registered port names or send to named ports
840that might or might not be persistent. Instead, you normally spawn a port
841on the remote node. The init function monitors the you, and you monitor
842the remote port. Since both monitors are local to the node, they are much
843more reliable.
844
845This also saves round-trips and avoids sending messages to the wrong port
846(hard to do in Erlang).
847
706=back 848=back
707 849
708=head1 SEE ALSO 850=head1 SEE ALSO
709 851
710L<AnyEvent>. 852L<AnyEvent>.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines