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.34 by root, Wed Aug 5 23:50:46 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
37
38 # monitoring
39 mon $port, $cb->(@msg) # callback is invoked on death
40 mon $port, $otherport # kill otherport on abnormal death
41 mon $port, $otherport, @msg # send message on death
27 42
28=head1 DESCRIPTION 43=head1 DESCRIPTION
29 44
30This module (-family) implements a simple message passing framework. 45This module (-family) implements a simple message passing framework.
31 46
104 119
105our $VERSION = '0.1'; 120our $VERSION = '0.1';
106our @EXPORT = qw( 121our @EXPORT = qw(
107 NODE $NODE *SELF node_of _any_ 122 NODE $NODE *SELF node_of _any_
108 resolve_node initialise_node 123 resolve_node initialise_node
109 snd rcv mon kil reg psub 124 snd rcv mon kil reg psub spawn
110 port 125 port
111); 126);
112 127
113our $SELF; 128our $SELF;
114 129
297 $port 312 $port
298} 313}
299 314
300=item reg $port, $name 315=item reg $port, $name
301 316
302Registers the given port under the name C<$name>. If the name already 317=item reg $name
303exists 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.
304 321
305A port can only be registered under one well known name. 322A port can only be registered under one well known name.
306 323
307A port automatically becomes unregistered when it is killed. 324A port automatically becomes unregistered when it is killed.
308 325
309=cut 326=cut
310 327
311sub reg(@) { 328sub reg(@) {
312 my ($port, $name) = @_; 329 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
313 330
314 $REG{$name} = $port; 331 $REG{$_[0]} = $port;
315} 332}
316 333
317=item rcv $port, $callback->(@msg) 334=item rcv $port, $callback->(@msg)
318 335
319Replaces the callback on the specified miniport (after converting it to 336Replaces the callback on the specified miniport (after converting it to
324=item rcv $port, $smartmatch => $callback->(@msg), ... 341=item rcv $port, $smartmatch => $callback->(@msg), ...
325 342
326=item rcv $port, [$smartmatch...] => $callback->(@msg), ... 343=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
327 344
328Register callbacks to be called on matching messages on the given full 345Register callbacks to be called on matching messages on the given full
329port (after converting it to one if required). 346port (after converting it to one if required) and return the port.
330 347
331The 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
332which 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
333registered. 350registered.
334 351
335The global C<$SELF> (exported by this module) contains C<$port> while 352The global C<$SELF> (exported by this module) contains C<$port> while
336executing the callback. 353executing the callback.
337 354
338Runtime errors wdurign callback execution will result in the port being 355Runtime errors during callback execution will result in the port being
339C<kil>ed. 356C<kil>ed.
340 357
341If 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
342first elements of the message, otherwise only the first element is being 359first elements of the message, otherwise only the first element is being
343matched. 360matched.
346exported by this module) matches any single element of the message. 363exported by this module) matches any single element of the message.
347 364
348While not required, it is highly recommended that the first matching 365While not required, it is highly recommended that the first matching
349element 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
350also 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 ;
351 384
352=cut 385=cut
353 386
354sub rcv($@) { 387sub rcv($@) {
355 my $port = shift; 388 my $port = shift;
462 } 495 }
463} 496}
464 497
465=item $guard = mon $port, $cb->(@reason) 498=item $guard = mon $port, $cb->(@reason)
466 499
467=item $guard = mon $port, $otherport 500=item $guard = mon $port, $rcvport
468 501
502=item $guard = mon $port
503
469=item $guard = mon $port, $otherport, @msg 504=item $guard = mon $port, $rcvport, @msg
470 505
471Monitor 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.
472 508
473In the first form, the callback is simply called with any number 509In the first form (callback), the callback is simply called with any
474of C<@reason> elements (no @reason means that the port was deleted 510number of C<@reason> elements (no @reason means that the port was deleted
475"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
476C<eval> if unsure. 512C<eval> if unsure.
477 513
478In 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)
479a @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
480under 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.
481 518
519The third form (kill self) is the same as the second form, except that
520C<$rvport> defaults to C<$SELF>.
521
482In 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.
483 531
484Example: call a given callback when C<$port> is killed. 532Example: call a given callback when C<$port> is killed.
485 533
486 mon $port, sub { warn "port died because of <@_>\n" }; 534 mon $port, sub { warn "port died because of <@_>\n" };
487 535
488Example: kill ourselves when C<$port> is killed abnormally. 536Example: kill ourselves when C<$port> is killed abnormally.
489 537
490 mon $port, $self; 538 mon $port;
491 539
492Example: send us a restart message another C<$port> is killed. 540Example: send us a restart message when another C<$port> is killed.
493 541
494 mon $port, $self => "restart"; 542 mon $port, $self => "restart";
495 543
496=cut 544=cut
497 545
498sub mon { 546sub mon {
499 my ($noderef, $port) = split /#/, shift, 2; 547 my ($noderef, $port) = split /#/, shift, 2;
500 548
501 my $node = $NODE{$noderef} || add_node $noderef; 549 my $node = $NODE{$noderef} || add_node $noderef;
502 550
503 my $cb = shift; 551 my $cb = @_ ? $_[0] : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
504 552
505 unless (ref $cb) { 553 unless (ref $cb) {
506 if (@_) { 554 if (@_) {
507 # send a kill info message 555 # send a kill info message
508 my (@msg) = ($cb, @_); 556 my (@msg) = @_;
509 $cb = sub { snd @msg, @_ }; 557 $cb = sub { snd @msg, @_ };
510 } else { 558 } else {
511 # simply kill other port 559 # simply kill other port
512 my $port = $cb; 560 my $port = $cb;
513 $cb = sub { kil $port, @_ if @_ }; 561 $cb = sub { kil $port, @_ if @_ };
539=cut 587=cut
540 588
541sub mon_guard { 589sub mon_guard {
542 my ($port, @refs) = @_; 590 my ($port, @refs) = @_;
543 591
592 #TODO: mon-less form?
593
544 mon $port, sub { 0 && @refs } 594 mon $port, sub { 0 && @refs }
545} 595}
546 596
547=item lnk $port1, $port2
548
549Link two ports. This is simply a shorthand for:
550
551 mon $port1, $port2;
552 mon $port2, $port1;
553
554It means that if either one is killed abnormally, the other one gets
555killed as well.
556
557=item kil $port[, @reason] 597=item kil $port[, @reason]
558 598
559Kill the specified port with the given C<@reason>. 599Kill the specified port with the given C<@reason>.
560 600
561If 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
567Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 607Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
568will be reported as reason C<< die => $@ >>. 608will be reported as reason C<< die => $@ >>.
569 609
570Transport/communication errors are reported as C<< transport_error => 610Transport/communication errors are reported as C<< transport_error =>
571$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}
572 683
573=back 684=back
574 685
575=head1 NODE MESSAGES 686=head1 NODE MESSAGES
576 687
618 729
619=back 730=back
620 731
621=head1 AnyEvent::MP vs. Distributed Erlang 732=head1 AnyEvent::MP vs. Distributed Erlang
622 733
623AnyEvent::MP got lots of its ideas from distributed erlang (erlang node 734AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
624== aemp node, erlang process == aemp port), so many of the documents and 735== aemp node, Erlang process == aemp port), so many of the documents and
625programming techniques employed by erlang apply to AnyEvent::MP. Here is a 736programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
626sample: 737sample:
627 738
628 http://www.erlang.se/doc/programming_rules.shtml 739 http://www.Erlang.se/doc/programming_rules.shtml
629 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4 740 http://Erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
630 http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6 741 http://Erlang.org/download/Erlang-book-part1.pdf # chapters 5 and 6
631 http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5 742 http://Erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
632 743
633Despite the similarities, there are also some important differences: 744Despite the similarities, there are also some important differences:
634 745
635=over 4 746=over 4
636 747
647 758
648Erlang uses processes that selctively receive messages, and therefore 759Erlang uses processes that selctively receive messages, and therefore
649needs a queue. AEMP is event based, queuing messages would serve no useful 760needs a queue. AEMP is event based, queuing messages would serve no useful
650purpose. 761purpose.
651 762
652(But see L<Coro::MP> for a more erlang-like process model on top of AEMP). 763(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
653 764
654=item * Erlang sends are synchronous, AEMP sends are asynchronous. 765=item * Erlang sends are synchronous, AEMP sends are asynchronous.
655 766
656Sending messages in erlang is synchronous and blocks the process. AEMP 767Sending messages in Erlang is synchronous and blocks the process. AEMP
657sends are immediate, connection establishment is handled in the 768sends are immediate, connection establishment is handled in the
658background. 769background.
659 770
660=item * Erlang can silently lose messages, AEMP cannot. 771=item * Erlang can silently lose messages, AEMP cannot.
661 772
664and c, and the other side only receives messages a and c). 775and c, and the other side only receives messages a and c).
665 776
666AEMP guarantees correct ordering, and the guarantee that there are no 777AEMP guarantees correct ordering, and the guarantee that there are no
667holes in the message sequence. 778holes in the message sequence.
668 779
669=item * In erlang, processes can be declared dead and later be found to be 780=item * In Erlang, processes can be declared dead and later be found to be
670alive. 781alive.
671 782
672In erlang it can happen that a monitored process is declared dead and 783In Erlang it can happen that a monitored process is declared dead and
673linked processes get killed, but later it turns out that the process is 784linked processes get killed, but later it turns out that the process is
674still alive - and can receive messages. 785still alive - and can receive messages.
675 786
676In AEMP, when port monitoring detects a port as dead, then that port will 787In AEMP, when port monitoring detects a port as dead, then that port will
677eventually be killed - it cannot happen that a node detects a port as dead 788eventually be killed - it cannot happen that a node detects a port as dead
678and then later sends messages to it, finding it is still alive. 789and then later sends messages to it, finding it is still alive.
679 790
680=item * Erlang can send messages to the wrong port, AEMP does not. 791=item * Erlang can send messages to the wrong port, AEMP does not.
681 792
682In erlang it is quite possible that a node that restarts reuses a process 793In Erlang it is quite possible that a node that restarts reuses a process
683ID known to other nodes for a completely different process, causing 794ID known to other nodes for a completely different process, causing
684messages destined for that process to end up in an unrelated process. 795messages destined for that process to end up in an unrelated process.
685 796
686AEMP never reuses port IDs, so old messages or old port IDs floating 797AEMP never reuses port IDs, so old messages or old port IDs floating
687around in the network will not be sent to an unrelated port. 798around in the network will not be sent to an unrelated port.
693securely authenticate nodes. 804securely authenticate nodes.
694 805
695=item * The AEMP protocol is optimised for both text-based and binary 806=item * The AEMP protocol is optimised for both text-based and binary
696communications. 807communications.
697 808
698The AEMP protocol, unlike the erlang protocol, supports both 809The AEMP protocol, unlike the Erlang protocol, supports both
699language-independent text-only protocols (good for debugging) and binary, 810language-independent text-only protocols (good for debugging) and binary,
700language-specific serialisers (e.g. Storable). 811language-specific serialisers (e.g. Storable).
701 812
702It has also been carefully designed to be implementable in other languages 813It has also been carefully designed to be implementable in other languages
703with a minimum of work while gracefully degrading fucntionality to make the 814with a minimum of work while gracefully degrading fucntionality to make the
704protocol simple. 815protocol simple.
705 816
817=item * AEMP has more flexible monitoring options than Erlang.
818
819In Erlang, you can chose to receive I<all> exit signals as messages
820or I<none>, there is no in-between, so monitoring single processes is
821difficult to implement. Monitoring in AEMP is more flexible than in
822Erlang, as one can choose between automatic kill, exit message or callback
823on a per-process basis.
824
825=item * Erlang tries to hide remote/local connections, AEMP does not.
826
827Monitoring in Erlang is not an indicator of process death/crashes,
828as linking is (except linking is unreliable in Erlang).
829
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).
838
706=back 839=back
707 840
708=head1 SEE ALSO 841=head1 SEE ALSO
709 842
710L<AnyEvent>. 843L<AnyEvent>.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines