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.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
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>).
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}
572 682
573=back 683=back
574 684
575=head1 NODE MESSAGES 685=head1 NODE MESSAGES
576 686
618 728
619=back 729=back
620 730
621=head1 AnyEvent::MP vs. Distributed Erlang 731=head1 AnyEvent::MP vs. Distributed Erlang
622 732
623AnyEvent::MP got lots of its ideas from distributed erlang (erlang node 733AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
624== aemp node, erlang process == aemp port), so many of the documents and 734== aemp node, Erlang process == aemp port), so many of the documents and
625programming techniques employed by erlang apply to AnyEvent::MP. Here is a 735programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
626sample: 736sample:
627 737
628 http://www.erlang.se/doc/programming_rules.shtml 738 http://www.Erlang.se/doc/programming_rules.shtml
629 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4 739 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 740 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 741 http://Erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
632 742
633Despite the similarities, there are also some important differences: 743Despite the similarities, there are also some important differences:
634 744
635=over 4 745=over 4
636 746
647 757
648Erlang uses processes that selctively receive messages, and therefore 758Erlang uses processes that selctively receive messages, and therefore
649needs a queue. AEMP is event based, queuing messages would serve no useful 759needs a queue. AEMP is event based, queuing messages would serve no useful
650purpose. 760purpose.
651 761
652(But see L<Coro::MP> for a more erlang-like process model on top of AEMP). 762(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
653 763
654=item * Erlang sends are synchronous, AEMP sends are asynchronous. 764=item * Erlang sends are synchronous, AEMP sends are asynchronous.
655 765
656Sending messages in erlang is synchronous and blocks the process. AEMP 766Sending messages in Erlang is synchronous and blocks the process. AEMP
657sends are immediate, connection establishment is handled in the 767sends are immediate, connection establishment is handled in the
658background. 768background.
659 769
660=item * Erlang can silently lose messages, AEMP cannot. 770=item * Erlang can silently lose messages, AEMP cannot.
661 771
664and c, and the other side only receives messages a and c). 774and c, and the other side only receives messages a and c).
665 775
666AEMP guarantees correct ordering, and the guarantee that there are no 776AEMP guarantees correct ordering, and the guarantee that there are no
667holes in the message sequence. 777holes in the message sequence.
668 778
669=item * In erlang, processes can be declared dead and later be found to be 779=item * In Erlang, processes can be declared dead and later be found to be
670alive. 780alive.
671 781
672In erlang it can happen that a monitored process is declared dead and 782In 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 783linked processes get killed, but later it turns out that the process is
674still alive - and can receive messages. 784still alive - and can receive messages.
675 785
676In AEMP, when port monitoring detects a port as dead, then that port will 786In 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 787eventually 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. 788and then later sends messages to it, finding it is still alive.
679 789
680=item * Erlang can send messages to the wrong port, AEMP does not. 790=item * Erlang can send messages to the wrong port, AEMP does not.
681 791
682In erlang it is quite possible that a node that restarts reuses a process 792In Erlang it is quite possible that a node that restarts reuses a process
683ID known to other nodes for a completely different process, causing 793ID known to other nodes for a completely different process, causing
684messages destined for that process to end up in an unrelated process. 794messages destined for that process to end up in an unrelated process.
685 795
686AEMP never reuses port IDs, so old messages or old port IDs floating 796AEMP never reuses port IDs, so old messages or old port IDs floating
687around in the network will not be sent to an unrelated port. 797around in the network will not be sent to an unrelated port.
693securely authenticate nodes. 803securely authenticate nodes.
694 804
695=item * The AEMP protocol is optimised for both text-based and binary 805=item * The AEMP protocol is optimised for both text-based and binary
696communications. 806communications.
697 807
698The AEMP protocol, unlike the erlang protocol, supports both 808The AEMP protocol, unlike the Erlang protocol, supports both
699language-independent text-only protocols (good for debugging) and binary, 809language-independent text-only protocols (good for debugging) and binary,
700language-specific serialisers (e.g. Storable). 810language-specific serialisers (e.g. Storable).
701 811
702It has also been carefully designed to be implementable in other languages 812It has also been carefully designed to be implementable in other languages
703with a minimum of work while gracefully degrading fucntionality to make the 813with a minimum of work while gracefully degrading fucntionality to make the
704protocol simple. 814protocol simple.
705 815
816=item * AEMP has more flexible monitoring options than Erlang.
817
818In Erlang, you can chose to receive I<all> exit signals as messages
819or I<none>, there is no in-between, so monitoring single processes is
820difficult to implement. Monitoring in AEMP is more flexible than in
821Erlang, as one can choose between automatic kill, exit message or callback
822on a per-process basis.
823
824=item * Erlang tries to hide remote/local connections, AEMP does not.
825
826Monitoring in Erlang is not an indicator of process death/crashes,
827as linking is (except linking is unreliable in Erlang).
828
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).
837
706=back 838=back
707 839
708=head1 SEE ALSO 840=head1 SEE ALSO
709 841
710L<AnyEvent>. 842L<AnyEvent>.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines