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.40 by root, Sat Aug 8 00:22: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
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>). To specify a function in the main
626program, use C<::name>.
627
628If the function doesn't exist, then the node tries to C<require>
629the package, then the package above the package and so on (e.g.
630C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
631exists or it runs out of package names.
632
633The init function is then called with the newly-created port as context
634object (C<$SELF>) and the C<@initdata> values as arguments.
635
636A common idiom is to pass your own port, monitor the spawned port, and
637in the init function, monitor the original port. This two-way monitoring
638ensures that both ports get cleaned up when there is a problem.
639
640Example: spawn a chat server port on C<$othernode>.
641
642 # this node, executed from within a port context:
643 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
644 mon $server;
645
646 # init function on C<$othernode>
647 sub connect {
648 my ($srcport) = @_;
649
650 mon $srcport;
651
652 rcv $SELF, sub {
653 ...
654 };
655 }
656
657=cut
658
659sub _spawn {
660 my $port = shift;
661 my $init = shift;
662
663 local $SELF = "$NODE#$port";
664 eval {
665 &{ load_func $init }
666 };
667 _self_die if $@;
668}
669
670sub spawn(@) {
671 my ($noderef, undef) = split /#/, shift, 2;
672
673 my $id = "$RUNIQ." . $ID++;
674
675 $_[0] =~ /::/
676 or Carp::croak "spawn init function must be a fully-qualified name, caught";
677
678 ($NODE{$noderef} || add_node $noderef)
679 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
680
681 "$noderef#$id"
682}
580 683
581=back 684=back
582 685
583=head1 NODE MESSAGES 686=head1 NODE MESSAGES
584 687
717or I<none>, there is no in-between, so monitoring single processes is 820or I<none>, there is no in-between, so monitoring single processes is
718difficult to implement. Monitoring in AEMP is more flexible than in 821difficult to implement. Monitoring in AEMP is more flexible than in
719Erlang, as one can choose between automatic kill, exit message or callback 822Erlang, as one can choose between automatic kill, exit message or callback
720on a per-process basis. 823on a per-process basis.
721 824
722=item * Erlang has different semantics for monitoring and linking, AEMP has the same. 825=item * Erlang tries to hide remote/local connections, AEMP does not.
723 826
724Monitoring in Erlang is not an indicator of process death/crashes, 827Monitoring in Erlang is not an indicator of process death/crashes,
725as linking is (except linking is unreliable in Erlang). In AEMP, the 828as linking is (except linking is unreliable in Erlang).
726semantics of monitoring and linking are identical, linking is simply 829
727two-way monitoring with automatic kill. 830In AEMP, you don't "look up" registered port names or send to named ports
831that might or might not be persistent. Instead, you normally spawn a port
832on the remote node. The init function monitors the you, and you monitor
833the remote port. Since both monitors are local to the node, they are much
834more reliable.
835
836This also saves round-trips and avoids sending messages to the wrong port
837(hard to do in Erlang).
728 838
729=back 839=back
730 840
731=head1 SEE ALSO 841=head1 SEE ALSO
732 842

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines