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.35 by root, Thu Aug 6 10:21: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
305 $port 312 $port
306} 313}
307 314
308=item reg $port, $name 315=item reg $port, $name
309 316
310Registers the given port under the name C<$name>. If the name already 317=item reg $name
311exists 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.
312 321
313A port can only be registered under one well known name. 322A port can only be registered under one well known name.
314 323
315A port automatically becomes unregistered when it is killed. 324A port automatically becomes unregistered when it is killed.
316 325
317=cut 326=cut
318 327
319sub reg(@) { 328sub reg(@) {
320 my ($port, $name) = @_; 329 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
321 330
322 $REG{$name} = $port; 331 $REG{$_[0]} = $port;
323} 332}
324 333
325=item rcv $port, $callback->(@msg) 334=item rcv $port, $callback->(@msg)
326 335
327Replaces the callback on the specified miniport (after converting it to 336Replaces the callback on the specified miniport (after converting it to
332=item rcv $port, $smartmatch => $callback->(@msg), ... 341=item rcv $port, $smartmatch => $callback->(@msg), ...
333 342
334=item rcv $port, [$smartmatch...] => $callback->(@msg), ... 343=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
335 344
336Register callbacks to be called on matching messages on the given full 345Register callbacks to be called on matching messages on the given full
337port (after converting it to one if required). 346port (after converting it to one if required) and return the port.
338 347
339The 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
340which 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
341registered. 350registered.
342 351
343The global C<$SELF> (exported by this module) contains C<$port> while 352The global C<$SELF> (exported by this module) contains C<$port> while
344executing the callback. 353executing the callback.
345 354
346Runtime errors wdurign callback execution will result in the port being 355Runtime errors during callback execution will result in the port being
347C<kil>ed. 356C<kil>ed.
348 357
349If 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
350first elements of the message, otherwise only the first element is being 359first elements of the message, otherwise only the first element is being
351matched. 360matched.
354exported by this module) matches any single element of the message. 363exported by this module) matches any single element of the message.
355 364
356While not required, it is highly recommended that the first matching 365While not required, it is highly recommended that the first matching
357element 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
358also 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 ;
359 384
360=cut 385=cut
361 386
362sub rcv($@) { 387sub rcv($@) {
363 my $port = shift; 388 my $port = shift;
470 } 495 }
471} 496}
472 497
473=item $guard = mon $port, $cb->(@reason) 498=item $guard = mon $port, $cb->(@reason)
474 499
475=item $guard = mon $port, $otherport 500=item $guard = mon $port, $rcvport
476 501
502=item $guard = mon $port
503
477=item $guard = mon $port, $otherport, @msg 504=item $guard = mon $port, $rcvport, @msg
478 505
479Monitor 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.
480 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
481In the first form, the callback is simply called with any number 518In the first form (callback), the callback is simply called with any
482of C<@reason> elements (no @reason means that the port was deleted 519number of C<@reason> elements (no @reason means that the port was deleted
483"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
484C<eval> if unsure. 521C<eval> if unsure.
485 522
486In 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)
487a @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
488under 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.
489 527
528The third form (kill self) is the same as the second form, except that
529C<$rvport> defaults to C<$SELF>.
530
490In 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.
491 540
492Example: call a given callback when C<$port> is killed. 541Example: call a given callback when C<$port> is killed.
493 542
494 mon $port, sub { warn "port died because of <@_>\n" }; 543 mon $port, sub { warn "port died because of <@_>\n" };
495 544
496Example: kill ourselves when C<$port> is killed abnormally. 545Example: kill ourselves when C<$port> is killed abnormally.
497 546
498 mon $port, $self; 547 mon $port;
499 548
500Example: send us a restart message another C<$port> is killed. 549Example: send us a restart message when another C<$port> is killed.
501 550
502 mon $port, $self => "restart"; 551 mon $port, $self => "restart";
503 552
504=cut 553=cut
505 554
506sub mon { 555sub mon {
507 my ($noderef, $port) = split /#/, shift, 2; 556 my ($noderef, $port) = split /#/, shift, 2;
508 557
509 my $node = $NODE{$noderef} || add_node $noderef; 558 my $node = $NODE{$noderef} || add_node $noderef;
510 559
511 my $cb = shift; 560 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
512 561
513 unless (ref $cb) { 562 unless (ref $cb) {
514 if (@_) { 563 if (@_) {
515 # send a kill info message 564 # send a kill info message
516 my (@msg) = ($cb, @_); 565 my (@msg) = ($cb, @_);
547=cut 596=cut
548 597
549sub mon_guard { 598sub mon_guard {
550 my ($port, @refs) = @_; 599 my ($port, @refs) = @_;
551 600
601 #TODO: mon-less form?
602
552 mon $port, sub { 0 && @refs } 603 mon $port, sub { 0 && @refs }
553} 604}
554 605
555=item lnk $port1, $port2
556
557Link two ports. This is simply a shorthand for:
558
559 mon $port1, $port2;
560 mon $port2, $port1;
561
562It means that if either one is killed abnormally, the other one gets
563killed as well.
564
565=item kil $port[, @reason] 606=item kil $port[, @reason]
566 607
567Kill the specified port with the given C<@reason>. 608Kill the specified port with the given C<@reason>.
568 609
569If 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
575Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 616Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
576will be reported as reason C<< die => $@ >>. 617will be reported as reason C<< die => $@ >>.
577 618
578Transport/communication errors are reported as C<< transport_error => 619Transport/communication errors are reported as C<< transport_error =>
579$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}
580 692
581=back 693=back
582 694
583=head1 NODE MESSAGES 695=head1 NODE MESSAGES
584 696
717or 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
718difficult to implement. Monitoring in AEMP is more flexible than in 830difficult to implement. Monitoring in AEMP is more flexible than in
719Erlang, as one can choose between automatic kill, exit message or callback 831Erlang, as one can choose between automatic kill, exit message or callback
720on a per-process basis. 832on a per-process basis.
721 833
722=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.
723 835
724Monitoring in Erlang is not an indicator of process death/crashes, 836Monitoring in Erlang is not an indicator of process death/crashes,
725as linking is (except linking is unreliable in Erlang). In AEMP, the 837as linking is (except linking is unreliable in Erlang).
726semantics of monitoring and linking are identical, linking is simply 838
727two-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).
728 847
729=back 848=back
730 849
731=head1 SEE ALSO 850=head1 SEE ALSO
732 851

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines