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.39 by root, Fri Aug 7 23:21:48 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, and
507optionally return a guard that can be used to stop monitoring again.
480 508
481In the first form, the callback is simply called with any number 509In the first form (callback), the callback is simply called with any
482of C<@reason> elements (no @reason means that the port was deleted 510number 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 511"normally"). Note also that I<< the callback B<must> never die >>, so use
484C<eval> if unsure. 512C<eval> if unsure.
485 513
486In the second form, the other port will be C<kil>'ed with C<@reason>, iff 514In the second form (another port given), the other port (C<$rcvport)
487a @reason was specified, i.e. on "normal" kils nothing happens, while 515will 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. 516"normal" kils nothing happens, while under all other conditions, the other
517port is killed with the same reason.
489 518
519The third form (kill self) is the same as the second form, except that
520C<$rvport> defaults to C<$SELF>.
521
490In the last form, a message of the form C<@msg, @reason> will be C<snd>. 522In the last form (message), a message of the form C<@msg, @reason> will be
523C<snd>.
524
525As a rule of thumb, monitoring requests should always monitor a port from
526a local port (or callback). The reason is that kill messages might get
527lost, just like any other message. Another less obvious reason is that
528even monitoring requests can get lost (for exmaple, when the connection
529to the other node goes down permanently). When monitoring a port locally
530these problems do not exist.
491 531
492Example: call a given callback when C<$port> is killed. 532Example: call a given callback when C<$port> is killed.
493 533
494 mon $port, sub { warn "port died because of <@_>\n" }; 534 mon $port, sub { warn "port died because of <@_>\n" };
495 535
496Example: kill ourselves when C<$port> is killed abnormally. 536Example: kill ourselves when C<$port> is killed abnormally.
497 537
498 mon $port, $self; 538 mon $port;
499 539
500Example: send us a restart message another C<$port> is killed. 540Example: send us a restart message when another C<$port> is killed.
501 541
502 mon $port, $self => "restart"; 542 mon $port, $self => "restart";
503 543
504=cut 544=cut
505 545
506sub mon { 546sub mon {
507 my ($noderef, $port) = split /#/, shift, 2; 547 my ($noderef, $port) = split /#/, shift, 2;
508 548
509 my $node = $NODE{$noderef} || add_node $noderef; 549 my $node = $NODE{$noderef} || add_node $noderef;
510 550
511 my $cb = shift; 551 my $cb = @_ ? $_[0] : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
512 552
513 unless (ref $cb) { 553 unless (ref $cb) {
514 if (@_) { 554 if (@_) {
515 # send a kill info message 555 # send a kill info message
516 my (@msg) = ($cb, @_); 556 my (@msg) = @_;
517 $cb = sub { snd @msg, @_ }; 557 $cb = sub { snd @msg, @_ };
518 } else { 558 } else {
519 # simply kill other port 559 # simply kill other port
520 my $port = $cb; 560 my $port = $cb;
521 $cb = sub { kil $port, @_ if @_ }; 561 $cb = sub { kil $port, @_ if @_ };
547=cut 587=cut
548 588
549sub mon_guard { 589sub mon_guard {
550 my ($port, @refs) = @_; 590 my ($port, @refs) = @_;
551 591
592 #TODO: mon-less form?
593
552 mon $port, sub { 0 && @refs } 594 mon $port, sub { 0 && @refs }
553} 595}
554 596
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] 597=item kil $port[, @reason]
566 598
567Kill the specified port with the given C<@reason>. 599Kill the specified port with the given C<@reason>.
568 600
569If no C<@reason> is specified, then the port is killed "normally" (linked 601If no C<@reason> is specified, then the port is killed "normally" (linked
575Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 607Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
576will be reported as reason C<< die => $@ >>. 608will be reported as reason C<< die => $@ >>.
577 609
578Transport/communication errors are reported as C<< transport_error => 610Transport/communication errors are reported as C<< transport_error =>
579$message >>. 611$message >>.
612
613=cut
614
615=item $port = spawn $node, $initfunc[, @initdata]
616
617Creates a port on the node C<$node> (which can also be a port ID, in which
618case it's the node where that port resides).
619
620The port ID of the newly created port is return immediately, and it is
621permissible to immediately start sending messages or monitor the port.
622
623After the port has been created, the init function is
624called. This function must be a fully-qualified function name
625(e.g. C<MyApp::Chat::Server::init>).
626
627If the function doesn't exist, then the node tries to C<require>
628the package, then the package above the package and so on (e.g.
629C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
630exists or it runs out of package names.
631
632The init function is then called with the newly-created port as context
633object (C<$SELF>) and the C<@initdata> values as arguments.
634
635A common idiom is to pass your own port, monitor the spawned port, and
636in the init function, monitor the original port. This two-way monitoring
637ensures that both ports get cleaned up when there is a problem.
638
639Example: spawn a chat server port on C<$othernode>.
640
641 # this node, executed from within a port context:
642 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
643 mon $server;
644
645 # init function on C<$othernode>
646 sub connect {
647 my ($srcport) = @_;
648
649 mon $srcport;
650
651 rcv $SELF, sub {
652 ...
653 };
654 }
655
656=cut
657
658sub _spawn {
659 my $port = shift;
660 my $init = shift;
661
662 local $SELF = "$NODE#$port";
663 eval {
664 &{ load_func $init }
665 };
666 _self_die if $@;
667}
668
669sub spawn(@) {
670 my ($noderef, undef) = split /#/, shift, 2;
671
672 my $id = "$RUNIQ." . $ID++;
673
674 $_[0] =~ /::/
675 or Carp::croak "spawn init function must be a fully-qualified name, caught";
676
677 ($NODE{$noderef} || add_node $noderef)
678 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
679
680 "$noderef#$id"
681}
580 682
581=back 683=back
582 684
583=head1 NODE MESSAGES 685=head1 NODE MESSAGES
584 686
717or I<none>, there is no in-between, so monitoring single processes is 819or I<none>, there is no in-between, so monitoring single processes is
718difficult to implement. Monitoring in AEMP is more flexible than in 820difficult to implement. Monitoring in AEMP is more flexible than in
719Erlang, as one can choose between automatic kill, exit message or callback 821Erlang, as one can choose between automatic kill, exit message or callback
720on a per-process basis. 822on a per-process basis.
721 823
722=item * Erlang has different semantics for monitoring and linking, AEMP has the same. 824=item * Erlang tries to hide remote/local connections, AEMP does not.
723 825
724Monitoring in Erlang is not an indicator of process death/crashes, 826Monitoring in Erlang is not an indicator of process death/crashes,
725as linking is (except linking is unreliable in Erlang). In AEMP, the 827as linking is (except linking is unreliable in Erlang).
726semantics of monitoring and linking are identical, linking is simply 828
727two-way monitoring with automatic kill. 829In AEMP, you don't "look up" registered port names or send to named ports
830that might or might not be persistent. Instead, you normally spawn a port
831on the remote node. The init function monitors the you, and you monitor
832the remote port. Since both monitors are local to the node, they are much
833more reliable.
834
835This also saves round-trips and avoids sending messages to the wrong port
836(hard to do in Erlang).
728 837
729=back 838=back
730 839
731=head1 SEE ALSO 840=head1 SEE ALSO
732 841

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines