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.100 by root, Fri Oct 2 20:41:56 2009 UTC vs.
Revision 1.113 by root, Sat Apr 3 15:32:18 2010 UTC

… …
30 rcv $port, pong => sub { warn "pong received\n" }; 30 rcv $port, pong => sub { warn "pong received\n" };
31 31
32 # create a port on another node 32 # create a port on another node
33 my $port = spawn $node, $initfunc, @initdata; 33 my $port = spawn $node, $initfunc, @initdata;
34 34
35 # destroy a prot again
36 kil $port; # "normal" kill
37 kil $port, my_error => "everything is broken"; # error kill
38
35 # monitoring 39 # monitoring
36 mon $localport, $cb->(@msg) # callback is invoked on death 40 mon $localport, $cb->(@msg) # callback is invoked on death
37 mon $localport, $otherport # kill otherport on abnormal death 41 mon $localport, $otherport # kill otherport on abnormal death
38 mon $localport, $otherport, @msg # send message on death 42 mon $localport, $otherport, @msg # send message on death
43
44 # temporarily execute code in port context
45 peval $port, sub { die "kill the port!" };
46
47 # execute callbacks in $SELF port context
48 my $timer = AE::timer 1, 0, psub {
49 die "kill the port, delayed";
50 };
39 51
40=head1 CURRENT STATUS 52=head1 CURRENT STATUS
41 53
42 bin/aemp - stable. 54 bin/aemp - stable.
43 AnyEvent::MP - stable API, should work. 55 AnyEvent::MP - stable API, should work.
… …
143 155
144use AE (); 156use AE ();
145 157
146use base "Exporter"; 158use base "Exporter";
147 159
148our $VERSION = 1.21; 160our $VERSION = 1.28;
149 161
150our @EXPORT = qw( 162our @EXPORT = qw(
151 NODE $NODE *SELF node_of after 163 NODE $NODE *SELF node_of after
152 configure 164 configure
153 snd rcv mon mon_guard kil psub spawn cal 165 snd rcv mon mon_guard kil psub peval spawn cal
154 port 166 port
155); 167);
156 168
157our $SELF; 169our $SELF;
158 170
… …
178 190
179Before a node can talk to other nodes on the network (i.e. enter 191Before a node can talk to other nodes on the network (i.e. enter
180"distributed mode") it has to configure itself - the minimum a node needs 192"distributed mode") it has to configure itself - the minimum a node needs
181to know is its own name, and optionally it should know the addresses of 193to know is its own name, and optionally it should know the addresses of
182some other nodes in the network to discover other nodes. 194some other nodes in the network to discover other nodes.
195
196The key/value pairs are basically the same ones as documented for the
197F<aemp> command line utility (sans the set/del prefix).
183 198
184This function configures a node - it must be called exactly once (or 199This function configures a node - it must be called exactly once (or
185never) before calling other AnyEvent::MP functions. 200never) before calling other AnyEvent::MP functions.
186 201
187=over 4 202=over 4
… …
371 msg1 => sub { ... }, 386 msg1 => sub { ... },
372 ... 387 ...
373 ; 388 ;
374 389
375Example: temporarily register a rcv callback for a tag matching some port 390Example: temporarily register a rcv callback for a tag matching some port
376(e.g. for a rpc reply) and unregister it after a message was received. 391(e.g. for an rpc reply) and unregister it after a message was received.
377 392
378 rcv $port, $otherport => sub { 393 rcv $port, $otherport => sub {
379 my @reply = @_; 394 my @reply = @_;
380 395
381 rcv $SELF, $otherport; 396 rcv $SELF, $otherport;
… …
394 if (ref $_[0]) { 409 if (ref $_[0]) {
395 if (my $self = $PORT_DATA{$portid}) { 410 if (my $self = $PORT_DATA{$portid}) {
396 "AnyEvent::MP::Port" eq ref $self 411 "AnyEvent::MP::Port" eq ref $self
397 or Carp::croak "$port: rcv can only be called on message matching ports, caught"; 412 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
398 413
399 $self->[2] = shift; 414 $self->[0] = shift;
400 } else { 415 } else {
401 my $cb = shift; 416 my $cb = shift;
402 $PORT{$portid} = sub { 417 $PORT{$portid} = sub {
403 local $SELF = $port; 418 local $SELF = $port;
404 eval { &$cb }; _self_die if $@; 419 eval { &$cb }; _self_die if $@;
405 }; 420 };
406 } 421 }
407 } elsif (defined $_[0]) { 422 } elsif (defined $_[0]) {
408 my $self = $PORT_DATA{$portid} ||= do { 423 my $self = $PORT_DATA{$portid} ||= do {
409 my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port"; 424 my $self = bless [$PORT{$portid} || sub { }, { }, $port], "AnyEvent::MP::Port";
410 425
411 $PORT{$portid} = sub { 426 $PORT{$portid} = sub {
412 local $SELF = $port; 427 local $SELF = $port;
413 428
414 if (my $cb = $self->[1]{$_[0]}) { 429 if (my $cb = $self->[1]{$_[0]}) {
… …
436 } 451 }
437 452
438 $port 453 $port
439} 454}
440 455
456=item peval $port, $coderef[, @args]
457
458Evaluates the given C<$codref> within the contetx of C<$port>, that is,
459when the code throews an exception the C<$port> will be killed.
460
461Any remaining args will be passed to the callback. Any return values will
462be returned to the caller.
463
464This is useful when you temporarily want to execute code in the context of
465a port.
466
467Example: create a port and run some initialisation code in it's context.
468
469 my $port = port { ... };
470
471 peval $port, sub {
472 init
473 or die "unable to init";
474 };
475
476=cut
477
478sub peval($$) {
479 local $SELF = shift;
480 my $cb = shift;
481
482 if (wantarray) {
483 my @res = eval { &$cb };
484 _self_die if $@;
485 @res
486 } else {
487 my $res = eval { &$cb };
488 _self_die if $@;
489 $res
490 }
491}
492
441=item $closure = psub { BLOCK } 493=item $closure = psub { BLOCK }
442 494
443Remembers C<$SELF> and creates a closure out of the BLOCK. When the 495Remembers C<$SELF> and creates a closure out of the BLOCK. When the
444closure is executed, sets up the environment in the same way as in C<rcv> 496closure is executed, sets up the environment in the same way as in C<rcv>
445callbacks, i.e. runtime errors will cause the port to get C<kil>ed. 497callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
498
499The effect is basically as if it returned C<< sub { peval $SELF, sub {
500BLOCK } } >>.
446 501
447This is useful when you register callbacks from C<rcv> callbacks: 502This is useful when you register callbacks from C<rcv> callbacks:
448 503
449 rcv delayed_reply => sub { 504 rcv delayed_reply => sub {
450 my ($delay, @reply) = @_; 505 my ($delay, @reply) = @_;
… …
598 653
599=item kil $port[, @reason] 654=item kil $port[, @reason]
600 655
601Kill the specified port with the given C<@reason>. 656Kill the specified port with the given C<@reason>.
602 657
603If no C<@reason> is specified, then the port is killed "normally" (ports 658If no C<@reason> is specified, then the port is killed "normally" -
604monitoring other ports will not necessarily die because a port dies 659monitor callback will be invoked, but the kil will not cause linked ports
605"normally"). 660(C<mon $mport, $lport> form) to get killed.
606 661
607Otherwise, linked ports get killed with the same reason (second form of 662If a C<@reason> is specified, then linked ports (C<mon $mport, $lport>
608C<mon>, see above). 663form) get killed with the same reason.
609 664
610Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 665Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
611will be reported as reason C<< die => $@ >>. 666will be reported as reason C<< die => $@ >>.
612 667
613Transport/communication errors are reported as C<< transport_error => 668Transport/communication errors are reported as C<< transport_error =>
… …
897overhead, as well as having to keep a proxy object everywhere. 952overhead, as well as having to keep a proxy object everywhere.
898 953
899Strings can easily be printed, easily serialised etc. and need no special 954Strings can easily be printed, easily serialised etc. and need no special
900procedures to be "valid". 955procedures to be "valid".
901 956
902And as a result, a miniport consists of a single closure stored in a 957And as a result, a port with just a default receiver consists of a single
903global hash - it can't become much cheaper. 958closure stored in a global hash - it can't become much cheaper.
904 959
905=item Why favour JSON, why not a real serialising format such as Storable? 960=item Why favour JSON, why not a real serialising format such as Storable?
906 961
907In fact, any AnyEvent::MP node will happily accept Storable as framing 962In fact, any AnyEvent::MP node will happily accept Storable as framing
908format, but currently there is no way to make a node use Storable by 963format, but currently there is no way to make a node use Storable by
… …
924 979
925L<AnyEvent::MP::Intro> - a gentle introduction. 980L<AnyEvent::MP::Intro> - a gentle introduction.
926 981
927L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 982L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
928 983
929L<AnyEvent::MP::Global> - network maintainance and port groups, to find 984L<AnyEvent::MP::Global> - network maintenance and port groups, to find
930your applications. 985your applications.
986
987L<AnyEvent::MP::DataConn> - establish data connections between nodes.
931 988
932L<AnyEvent::MP::LogCatcher> - simple service to display log messages from 989L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
933all nodes. 990all nodes.
934 991
935L<AnyEvent>. 992L<AnyEvent>.

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines