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.44 by root, Wed Aug 12 21:39:58 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
98 105
99=cut 106=cut
100 107
101package AnyEvent::MP; 108package AnyEvent::MP;
102 109
103use AnyEvent::MP::Base; 110use AnyEvent::MP::Kernel;
104 111
105use common::sense; 112use common::sense;
106 113
107use Carp (); 114use Carp ();
108 115
109use AE (); 116use AE ();
110 117
111use base "Exporter"; 118use base "Exporter";
112 119
113our $VERSION = '0.1'; 120our $VERSION = $AnyEvent::MP::Kernel::VERSION;
121
114our @EXPORT = qw( 122our @EXPORT = qw(
115 NODE $NODE *SELF node_of _any_ 123 NODE $NODE *SELF node_of _any_
116 resolve_node initialise_node 124 resolve_node initialise_node
117 snd rcv mon kil reg psub 125 snd rcv mon kil reg psub spawn
118 port 126 port
119); 127);
120 128
121our $SELF; 129our $SELF;
122 130
305 $port 313 $port
306} 314}
307 315
308=item reg $port, $name 316=item reg $port, $name
309 317
310Registers the given port under the name C<$name>. If the name already 318=item reg $name
311exists it is replaced. 319
320Registers the given port (or C<$SELF><<< if missing) under the name
321C<$name>. If the name already exists it is replaced.
312 322
313A port can only be registered under one well known name. 323A port can only be registered under one well known name.
314 324
315A port automatically becomes unregistered when it is killed. 325A port automatically becomes unregistered when it is killed.
316 326
317=cut 327=cut
318 328
319sub reg(@) { 329sub reg(@) {
320 my ($port, $name) = @_; 330 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
321 331
322 $REG{$name} = $port; 332 $REG{$_[0]} = $port;
323} 333}
324 334
325=item rcv $port, $callback->(@msg) 335=item rcv $port, $callback->(@msg)
326 336
327Replaces the callback on the specified miniport (after converting it to 337Replaces the callback on the specified miniport (after converting it to
332=item rcv $port, $smartmatch => $callback->(@msg), ... 342=item rcv $port, $smartmatch => $callback->(@msg), ...
333 343
334=item rcv $port, [$smartmatch...] => $callback->(@msg), ... 344=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
335 345
336Register callbacks to be called on matching messages on the given full 346Register callbacks to be called on matching messages on the given full
337port (after converting it to one if required). 347port (after converting it to one if required) and return the port.
338 348
339The callback has to return a true value when its work is done, after 349The 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 350which is will be removed, or a false value in which case it will stay
341registered. 351registered.
342 352
343The global C<$SELF> (exported by this module) contains C<$port> while 353The global C<$SELF> (exported by this module) contains C<$port> while
344executing the callback. 354executing the callback.
345 355
346Runtime errors wdurign callback execution will result in the port being 356Runtime errors during callback execution will result in the port being
347C<kil>ed. 357C<kil>ed.
348 358
349If the match is an array reference, then it will be matched against the 359If 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 360first elements of the message, otherwise only the first element is being
351matched. 361matched.
354exported by this module) matches any single element of the message. 364exported by this module) matches any single element of the message.
355 365
356While not required, it is highly recommended that the first matching 366While not required, it is highly recommended that the first matching
357element is a string identifying the message. The one-string-only match is 367element is a string identifying the message. The one-string-only match is
358also the most efficient match (by far). 368also the most efficient match (by far).
369
370Example: create a port and bind receivers on it in one go.
371
372 my $port = rcv port,
373 msg1 => sub { ...; 0 },
374 msg2 => sub { ...; 0 },
375 ;
376
377Example: create a port, bind receivers and send it in a message elsewhere
378in one go:
379
380 snd $otherport, reply =>
381 rcv port,
382 msg1 => sub { ...; 0 },
383 ...
384 ;
359 385
360=cut 386=cut
361 387
362sub rcv($@) { 388sub rcv($@) {
363 my $port = shift; 389 my $port = shift;
470 } 496 }
471} 497}
472 498
473=item $guard = mon $port, $cb->(@reason) 499=item $guard = mon $port, $cb->(@reason)
474 500
475=item $guard = mon $port, $otherport 501=item $guard = mon $port, $rcvport
476 502
503=item $guard = mon $port
504
477=item $guard = mon $port, $otherport, @msg 505=item $guard = mon $port, $rcvport, @msg
478 506
479Monitor the given port and do something when the port is killed. 507Monitor the given port and do something when the port is killed or
508messages to it were lost, and optionally return a guard that can be used
509to stop monitoring again.
480 510
511C<mon> effectively guarantees that, in the absence of hardware failures,
512that after starting the monitor, either all messages sent to the port
513will arrive, or the monitoring action will be invoked after possible
514message loss has been detected. No messages will be lost "in between"
515(after the first lost message no further messages will be received by the
516port). After the monitoring action was invoked, further messages might get
517delivered again.
518
481In the first form, the callback is simply called with any number 519In the first form (callback), the callback is simply called with any
482of C<@reason> elements (no @reason means that the port was deleted 520number 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 521"normally"). Note also that I<< the callback B<must> never die >>, so use
484C<eval> if unsure. 522C<eval> if unsure.
485 523
486In the second form, the other port will be C<kil>'ed with C<@reason>, iff 524In the second form (another port given), the other port (C<$rcvport>)
487a @reason was specified, i.e. on "normal" kils nothing happens, while 525will 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. 526"normal" kils nothing happens, while under all other conditions, the other
527port is killed with the same reason.
489 528
529The third form (kill self) is the same as the second form, except that
530C<$rvport> defaults to C<$SELF>.
531
490In the last form, a message of the form C<@msg, @reason> will be C<snd>. 532In the last form (message), a message of the form C<@msg, @reason> will be
533C<snd>.
534
535As a rule of thumb, monitoring requests should always monitor a port from
536a local port (or callback). The reason is that kill messages might get
537lost, just like any other message. Another less obvious reason is that
538even monitoring requests can get lost (for exmaple, when the connection
539to the other node goes down permanently). When monitoring a port locally
540these problems do not exist.
491 541
492Example: call a given callback when C<$port> is killed. 542Example: call a given callback when C<$port> is killed.
493 543
494 mon $port, sub { warn "port died because of <@_>\n" }; 544 mon $port, sub { warn "port died because of <@_>\n" };
495 545
496Example: kill ourselves when C<$port> is killed abnormally. 546Example: kill ourselves when C<$port> is killed abnormally.
497 547
498 mon $port, $self; 548 mon $port;
499 549
500Example: send us a restart message another C<$port> is killed. 550Example: send us a restart message when another C<$port> is killed.
501 551
502 mon $port, $self => "restart"; 552 mon $port, $self => "restart";
503 553
504=cut 554=cut
505 555
506sub mon { 556sub mon {
507 my ($noderef, $port) = split /#/, shift, 2; 557 my ($noderef, $port) = split /#/, shift, 2;
508 558
509 my $node = $NODE{$noderef} || add_node $noderef; 559 my $node = $NODE{$noderef} || add_node $noderef;
510 560
511 my $cb = shift; 561 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
512 562
513 unless (ref $cb) { 563 unless (ref $cb) {
514 if (@_) { 564 if (@_) {
515 # send a kill info message 565 # send a kill info message
516 my (@msg) = ($cb, @_); 566 my (@msg) = ($cb, @_);
547=cut 597=cut
548 598
549sub mon_guard { 599sub mon_guard {
550 my ($port, @refs) = @_; 600 my ($port, @refs) = @_;
551 601
602 #TODO: mon-less form?
603
552 mon $port, sub { 0 && @refs } 604 mon $port, sub { 0 && @refs }
553} 605}
554 606
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] 607=item kil $port[, @reason]
566 608
567Kill the specified port with the given C<@reason>. 609Kill the specified port with the given C<@reason>.
568 610
569If no C<@reason> is specified, then the port is killed "normally" (linked 611If no C<@reason> is specified, then the port is killed "normally" (linked
575Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 617Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
576will be reported as reason C<< die => $@ >>. 618will be reported as reason C<< die => $@ >>.
577 619
578Transport/communication errors are reported as C<< transport_error => 620Transport/communication errors are reported as C<< transport_error =>
579$message >>. 621$message >>.
622
623=cut
624
625=item $port = spawn $node, $initfunc[, @initdata]
626
627Creates a port on the node C<$node> (which can also be a port ID, in which
628case it's the node where that port resides).
629
630The port ID of the newly created port is return immediately, and it is
631permissible to immediately start sending messages or monitor the port.
632
633After the port has been created, the init function is
634called. This function must be a fully-qualified function name
635(e.g. C<MyApp::Chat::Server::init>). To specify a function in the main
636program, use C<::name>.
637
638If the function doesn't exist, then the node tries to C<require>
639the package, then the package above the package and so on (e.g.
640C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
641exists or it runs out of package names.
642
643The init function is then called with the newly-created port as context
644object (C<$SELF>) and the C<@initdata> values as arguments.
645
646A common idiom is to pass your own port, monitor the spawned port, and
647in the init function, monitor the original port. This two-way monitoring
648ensures that both ports get cleaned up when there is a problem.
649
650Example: spawn a chat server port on C<$othernode>.
651
652 # this node, executed from within a port context:
653 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
654 mon $server;
655
656 # init function on C<$othernode>
657 sub connect {
658 my ($srcport) = @_;
659
660 mon $srcport;
661
662 rcv $SELF, sub {
663 ...
664 };
665 }
666
667=cut
668
669sub _spawn {
670 my $port = shift;
671 my $init = shift;
672
673 local $SELF = "$NODE#$port";
674 eval {
675 &{ load_func $init }
676 };
677 _self_die if $@;
678}
679
680sub spawn(@) {
681 my ($noderef, undef) = split /#/, shift, 2;
682
683 my $id = "$RUNIQ." . $ID++;
684
685 $_[0] =~ /::/
686 or Carp::croak "spawn init function must be a fully-qualified name, caught";
687
688 ($NODE{$noderef} || add_node $noderef)
689 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
690
691 "$noderef#$id"
692}
580 693
581=back 694=back
582 695
583=head1 NODE MESSAGES 696=head1 NODE MESSAGES
584 697
717or I<none>, there is no in-between, so monitoring single processes is 830or I<none>, there is no in-between, so monitoring single processes is
718difficult to implement. Monitoring in AEMP is more flexible than in 831difficult to implement. Monitoring in AEMP is more flexible than in
719Erlang, as one can choose between automatic kill, exit message or callback 832Erlang, as one can choose between automatic kill, exit message or callback
720on a per-process basis. 833on a per-process basis.
721 834
722=item * Erlang has different semantics for monitoring and linking, AEMP has the same. 835=item * Erlang tries to hide remote/local connections, AEMP does not.
723 836
724Monitoring in Erlang is not an indicator of process death/crashes, 837Monitoring in Erlang is not an indicator of process death/crashes,
725as linking is (except linking is unreliable in Erlang). In AEMP, the 838as linking is (except linking is unreliable in Erlang).
726semantics of monitoring and linking are identical, linking is simply 839
727two-way monitoring with automatic kill. 840In AEMP, you don't "look up" registered port names or send to named ports
841that might or might not be persistent. Instead, you normally spawn a port
842on the remote node. The init function monitors the you, and you monitor
843the remote port. Since both monitors are local to the node, they are much
844more reliable.
845
846This also saves round-trips and avoids sending messages to the wrong port
847(hard to do in Erlang).
728 848
729=back 849=back
730 850
731=head1 SEE ALSO 851=head1 SEE ALSO
732 852

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines