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.36 by root, Thu Aug 6 10:46:48 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
27
28 # linking two ports, so they both crash together
29 lnk $port1, $port2;
30 37
31 # monitoring 38 # monitoring
32 mon $port, $cb->(@msg) # callback is invoked on death 39 mon $port, $cb->(@msg) # callback is invoked on death
33 mon $port, $otherport # kill otherport on abnormal death 40 mon $port, $otherport # kill otherport on abnormal death
34 mon $port, $otherport, @msg # send message on death 41 mon $port, $otherport, @msg # send message on death
… …
112 119
113our $VERSION = '0.1'; 120our $VERSION = '0.1';
114our @EXPORT = qw( 121our @EXPORT = qw(
115 NODE $NODE *SELF node_of _any_ 122 NODE $NODE *SELF node_of _any_
116 resolve_node initialise_node 123 resolve_node initialise_node
117 snd rcv mon kil reg psub 124 snd rcv mon kil reg psub spawn
118 port 125 port
119); 126);
120 127
121our $SELF; 128our $SELF;
122 129
… …
343registered. 350registered.
344 351
345The global C<$SELF> (exported by this module) contains C<$port> while 352The global C<$SELF> (exported by this module) contains C<$port> while
346executing the callback. 353executing the callback.
347 354
348Runtime errors wdurign callback execution will result in the port being 355Runtime errors during callback execution will result in the port being
349C<kil>ed. 356C<kil>ed.
350 357
351If 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
352first elements of the message, otherwise only the first element is being 359first elements of the message, otherwise only the first element is being
353matched. 360matched.
… …
494 501
495=item $guard = mon $port 502=item $guard = mon $port
496 503
497=item $guard = mon $port, $rcvport, @msg 504=item $guard = mon $port, $rcvport, @msg
498 505
499Monitor the given port and do something when the port is killed, and 506Monitor the given port and do something when the port is killed or
500optionally return a guard that can be used to stop monitoring again. 507messages to it were lost, and optionally return a guard that can be used
508to stop monitoring again.
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.
501 517
502In the first form (callback), the callback is simply called with any 518In the first form (callback), the callback is simply called with any
503number of C<@reason> elements (no @reason means that the port was deleted 519number of C<@reason> elements (no @reason means that the port was deleted
504"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
505C<eval> if unsure. 521C<eval> if unsure.
… …
513C<$rvport> defaults to C<$SELF>. 529C<$rvport> defaults to C<$SELF>.
514 530
515In the last form (message), a message of the form C<@msg, @reason> will be 531In the last form (message), a message of the form C<@msg, @reason> will be
516C<snd>. 532C<snd>.
517 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.
540
518Example: call a given callback when C<$port> is killed. 541Example: call a given callback when C<$port> is killed.
519 542
520 mon $port, sub { warn "port died because of <@_>\n" }; 543 mon $port, sub { warn "port died because of <@_>\n" };
521 544
522Example: kill ourselves when C<$port> is killed abnormally. 545Example: kill ourselves when C<$port> is killed abnormally.
… …
532sub mon { 555sub mon {
533 my ($noderef, $port) = split /#/, shift, 2; 556 my ($noderef, $port) = split /#/, shift, 2;
534 557
535 my $node = $NODE{$noderef} || add_node $noderef; 558 my $node = $NODE{$noderef} || add_node $noderef;
536 559
537 my $cb = @_ ? $_[0] : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,'; 560 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
538 561
539 unless (ref $cb) { 562 unless (ref $cb) {
540 if (@_) { 563 if (@_) {
541 # send a kill info message 564 # send a kill info message
542 my (@msg) = @_; 565 my (@msg) = ($cb, @_);
543 $cb = sub { snd @msg, @_ }; 566 $cb = sub { snd @msg, @_ };
544 } else { 567 } else {
545 # simply kill other port 568 # simply kill other port
546 my $port = $cb; 569 my $port = $cb;
547 $cb = sub { kil $port, @_ if @_ }; 570 $cb = sub { kil $port, @_ if @_ };
… …
578 #TODO: mon-less form? 601 #TODO: mon-less form?
579 602
580 mon $port, sub { 0 && @refs } 603 mon $port, sub { 0 && @refs }
581} 604}
582 605
583=item lnk $port1, $port2
584
585=item lnk $otherport
586
587Link two ports. This is simply a shorthand for:
588
589 mon $port1, $port2;
590 mon $port2, $port1;
591
592It means that if either one is killed abnormally, the other one gets
593killed as well.
594
595The one-argument form assumes that one port is C<$SELF>.
596
597=cut
598
599sub lnk {
600 my $port1 = shift;
601 my $port2 = @_ ? shift : $SELF || Carp::croak 'lnk: called with one argument only, but $SELF not set,';
602
603 mon $port1, $port2;
604 mon $port2, $port1;
605}
606
607=item kil $port[, @reason] 606=item kil $port[, @reason]
608 607
609Kill the specified port with the given C<@reason>. 608Kill the specified port with the given C<@reason>.
610 609
611If 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
… …
617Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 616Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
618will be reported as reason C<< die => $@ >>. 617will be reported as reason C<< die => $@ >>.
619 618
620Transport/communication errors are reported as C<< transport_error => 619Transport/communication errors are reported as C<< transport_error =>
621$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}
622 692
623=back 693=back
624 694
625=head1 NODE MESSAGES 695=head1 NODE MESSAGES
626 696
… …
759or I<none>, there is no in-between, so monitoring single processes is 829or I<none>, there is no in-between, so monitoring single processes is
760difficult to implement. Monitoring in AEMP is more flexible than in 830difficult to implement. Monitoring in AEMP is more flexible than in
761Erlang, as one can choose between automatic kill, exit message or callback 831Erlang, as one can choose between automatic kill, exit message or callback
762on a per-process basis. 832on a per-process basis.
763 833
764=item * Erlang has different semantics for monitoring and linking, AEMP has the same. 834=item * Erlang tries to hide remote/local connections, AEMP does not.
765 835
766Monitoring in Erlang is not an indicator of process death/crashes, 836Monitoring in Erlang is not an indicator of process death/crashes,
767as linking is (except linking is unreliable in Erlang). In AEMP, the 837as linking is (except linking is unreliable in Erlang).
768semantics of monitoring and linking are identical, linking is simply 838
769two-way monitoring with automatic kill. 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).
770 847
771=back 848=back
772 849
773=head1 SEE ALSO 850=head1 SEE ALSO
774 851

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines