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.84 by root, Tue Sep 8 01:42:14 2009 UTC vs.
Revision 1.141 by root, Fri Mar 23 03:24:41 2012 UTC

1=head1 NAME 1=head1 NAME
2 2
3AnyEvent::MP - multi-processing/message-passing framework 3AnyEvent::MP - erlang-style multi-processing/message-passing framework
4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::MP; 7 use AnyEvent::MP;
8 8
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 port again
36 kil $port; # "normal" kill
37 kil $port, my_error => "everything is broken"; # error kill
38
35 # monitoring 39 # monitoring
36 mon $port, $cb->(@msg) # callback is invoked on death 40 mon $port, $cb->(@msg) # callback is invoked on death
37 mon $port, $otherport # kill otherport on abnormal death 41 mon $port, $localport # kill localport on abnormal death
38 mon $port, $otherport, @msg # send message on death 42 mon $port, $localport, @msg # send message on death
39 43
40=head1 CURRENT STATUS 44 # temporarily execute code in port context
45 peval $port, sub { die "kill the port!" };
41 46
42 bin/aemp - stable. 47 # execute callbacks in $SELF port context
43 AnyEvent::MP - stable API, should work. 48 my $timer = AE::timer 1, 0, psub {
44 AnyEvent::MP::Intro - explains most concepts. 49 die "kill the port, delayed";
45 AnyEvent::MP::Kernel - mostly stable. 50 };
46 AnyEvent::MP::Global - stable but incomplete, protocol not yet final.
47
48stay tuned.
49 51
50=head1 DESCRIPTION 52=head1 DESCRIPTION
51 53
52This module (-family) implements a simple message passing framework. 54This module (-family) implements a simple message passing framework.
53 55
68 70
69Ports allow you to register C<rcv> handlers that can match all or just 71Ports allow you to register C<rcv> handlers that can match all or just
70some messages. Messages send to ports will not be queued, regardless of 72some messages. Messages send to ports will not be queued, regardless of
71anything was listening for them or not. 73anything was listening for them or not.
72 74
75Ports are represented by (printable) strings called "port IDs".
76
73=item port ID - C<nodeid#portname> 77=item port ID - C<nodeid#portname>
74 78
75A port ID is the concatenation of a node ID, a hash-mark (C<#>) as 79A port ID is the concatenation of a node ID, a hash-mark (C<#>)
76separator, and a port name (a printable string of unspecified format). 80as separator, and a port name (a printable string of unspecified
81format created by AnyEvent::MP).
77 82
78=item node 83=item node
79 84
80A node is a single process containing at least one port - the node port, 85A node is a single process containing at least one port - the node port,
81which enables nodes to manage each other remotely, and to create new 86which enables nodes to manage each other remotely, and to create new
82ports. 87ports.
83 88
84Nodes are either public (have one or more listening ports) or private 89Nodes are either public (have one or more listening ports) or private
85(no listening ports). Private nodes cannot talk to other private nodes 90(no listening ports). Private nodes cannot talk to other private nodes
86currently. 91currently, but all nodes can talk to public nodes.
87 92
93Nodes is represented by (printable) strings called "node IDs".
94
88=item node ID - C<[A-Z_][a-zA-Z0-9_\-.:]*> 95=item node ID - C<[A-Za-z0-9_\-.:]*>
89 96
90A node ID is a string that uniquely identifies the node within a 97A node ID is a string that uniquely identifies the node within a
91network. Depending on the configuration used, node IDs can look like a 98network. Depending on the configuration used, node IDs can look like a
92hostname, a hostname and a port, or a random string. AnyEvent::MP itself 99hostname, a hostname and a port, or a random string. AnyEvent::MP itself
93doesn't interpret node IDs in any way. 100doesn't interpret node IDs in any way except to uniquely identify a node.
94 101
95=item binds - C<ip:port> 102=item binds - C<ip:port>
96 103
97Nodes can only talk to each other by creating some kind of connection to 104Nodes can only talk to each other by creating some kind of connection to
98each other. To do this, nodes should listen on one or more local transport 105each other. To do this, nodes should listen on one or more local transport
106endpoints - binds.
107
99endpoints - binds. Currently, only standard C<ip:port> specifications can 108Currently, only standard C<ip:port> specifications can be used, which
100be used, which specify TCP ports to listen on. 109specify TCP ports to listen on. So a bind is basically just a tcp socket
110in listening mode thta accepts conenctions form other nodes.
101 111
102=item seed nodes 112=item seed nodes
103 113
104When a node starts, it knows nothing about the network. To teach the node 114When a node starts, it knows nothing about the network it is in - it
105about the network it first has to contact some other node within the 115needs to connect to at least one other node that is already in the
106network. This node is called a seed. 116network. These other nodes are called "seed nodes".
107 117
108Apart from the fact that other nodes know them as seed nodes and they have 118Seed nodes themselves are not special - they are seed nodes only because
109to have fixed listening addresses, seed nodes are perfectly normal nodes - 119some other node I<uses> them as such, but any node can be used as seed
110any node can function as a seed node for others. 120node for other nodes, and eahc node cna use a different set of seed nodes.
111 121
112In addition to discovering the network, seed nodes are also used to 122In addition to discovering the network, seed nodes are also used to
113maintain the network and to connect nodes that otherwise would have 123maintain the network - all nodes using the same seed node form are part of
114trouble connecting. They form the backbone of the AnyEvent::MP network. 124the same network. If a network is split into multiple subnets because e.g.
125the network link between the parts goes down, then using the same seed
126nodes for all nodes ensures that eventually the subnets get merged again.
115 127
116Seed nodes are expected to be long-running, and at least one seed node 128Seed nodes are expected to be long-running, and at least one seed node
117should always be available. 129should always be available. They should also be relatively responsive - a
130seed node that blocks for long periods will slow down everybody else.
118 131
132For small networks, it's best if every node uses the same set of seed
133nodes. For large networks, it can be useful to specify "regional" seed
134nodes for most nodes in an area, and use all seed nodes as seed nodes for
135each other. What's important is that all seed nodes connections form a
136complete graph, so that the network cannot split into separate subnets
137forever.
138
139Seed nodes are represented by seed IDs.
140
119=item seeds - C<host:port> 141=item seed IDs - C<host:port>
120 142
121Seeds are transport endpoint(s) (usually a hostname/IP address and a 143Seed IDs are transport endpoint(s) (usually a hostname/IP address and a
122TCP port) of nodes thta should be used as seed nodes. 144TCP port) of nodes that should be used as seed nodes.
123 145
124The nodes listening on those endpoints are expected to be long-running, 146=item global nodes
125and at least one of those should always be available. When nodes run out 147
126of connections (e.g. due to a network error), they try to re-establish 148An AEMP network needs a discovery service - nodes need to know how to
127connections to some seednodes again to join the network. 149connect to other nodes they only know by name. In addition, AEMP offers a
150distributed "group database", which maps group names to a list of strings
151- for example, to register worker ports.
152
153A network needs at least one global node to work, and allows every node to
154be a global node.
155
156Any node that loads the L<AnyEvent::MP::Global> module becomes a global
157node and tries to keep connections to all other nodes. So while it can
158make sense to make every node "global" in small networks, it usually makes
159sense to only make seed nodes into global nodes in large networks (nodes
160keep connections to seed nodes and global nodes, so makign them the same
161reduces overhead).
128 162
129=back 163=back
130 164
131=head1 VARIABLES/FUNCTIONS 165=head1 VARIABLES/FUNCTIONS
132 166
134 168
135=cut 169=cut
136 170
137package AnyEvent::MP; 171package AnyEvent::MP;
138 172
173use AnyEvent::MP::Config ();
139use AnyEvent::MP::Kernel; 174use AnyEvent::MP::Kernel;
175use AnyEvent::MP::Kernel qw(%NODE %PORT %PORT_DATA $UNIQ $RUNIQ $ID);
140 176
141use common::sense; 177use common::sense;
142 178
143use Carp (); 179use Carp ();
144 180
145use AE (); 181use AnyEvent ();
182use Guard ();
146 183
147use base "Exporter"; 184use base "Exporter";
148 185
149our $VERSION = $AnyEvent::MP::Kernel::VERSION; 186our $VERSION = $AnyEvent::MP::Config::VERSION;
150 187
151our @EXPORT = qw( 188our @EXPORT = qw(
152 NODE $NODE *SELF node_of after 189 NODE $NODE *SELF node_of after
153 configure 190 configure
154 snd rcv mon mon_guard kil reg psub spawn 191 snd rcv mon mon_guard kil psub peval spawn cal
155 port 192 port
193 db_set db_del db_reg
194 db_mon db_family db_keys db_values
156); 195);
157 196
158our $SELF; 197our $SELF;
159 198
160sub _self_die() { 199sub _self_die() {
183some other nodes in the network to discover other nodes. 222some other nodes in the network to discover other nodes.
184 223
185This function configures a node - it must be called exactly once (or 224This function configures a node - it must be called exactly once (or
186never) before calling other AnyEvent::MP functions. 225never) before calling other AnyEvent::MP functions.
187 226
227The key/value pairs are basically the same ones as documented for the
228F<aemp> command line utility (sans the set/del prefix), with these additions:
229
230=over 4
231
232=item norc => $boolean (default false)
233
234If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not>
235be consulted - all configuraiton options must be specified in the
236C<configure> call.
237
238=item force => $boolean (default false)
239
240IF true, then the values specified in the C<configure> will take
241precedence over any values configured via the rc file. The default is for
242the rc file to override any options specified in the program.
243
244=item secure => $pass->(@msg)
245
246In addition to specifying a boolean, you can specify a code reference that
247is called for every code execution attempt - the execution request is
248granted iff the callback returns a true value.
249
250Most of the time the callback should look only at
251C<$AnyEvent::MP::Kernel::SRCNODE> to make a decision, and not at the
252actual message (which can be about anything, and is mostly provided for
253diagnostic purposes).
254
255See F<semp setsecure> for more info.
256
257=back
258
188=over 4 259=over 4
189 260
190=item step 1, gathering configuration from profiles 261=item step 1, gathering configuration from profiles
191 262
192The function first looks up a profile in the aemp configuration (see the 263The function first looks up a profile in the aemp configuration (see the
205That means that the values specified in the profile have highest priority 276That means that the values specified in the profile have highest priority
206and the values specified directly via C<configure> have lowest priority, 277and the values specified directly via C<configure> have lowest priority,
207and can only be used to specify defaults. 278and can only be used to specify defaults.
208 279
209If the profile specifies a node ID, then this will become the node ID of 280If the profile specifies a node ID, then this will become the node ID of
210this process. If not, then the profile name will be used as node ID. The 281this process. If not, then the profile name will be used as node ID, with
211special node ID of C<anon/> will be replaced by a random node ID. 282a unique randoms tring (C</%u>) appended.
283
284The node ID can contain some C<%> sequences that are expanded: C<%n>
285is expanded to the local nodename, C<%u> is replaced by a random
286strign to make the node unique. For example, the F<aemp> commandline
287utility uses C<aemp/%n/%u> as nodename, which might expand to
288C<aemp/cerebro/ZQDGSIkRhEZQDGSIkRhE>.
212 289
213=item step 2, bind listener sockets 290=item step 2, bind listener sockets
214 291
215The next step is to look up the binds in the profile, followed by binding 292The next step is to look up the binds in the profile, followed by binding
216aemp protocol listeners on all binds specified (it is possible and valid 293aemp protocol listeners on all binds specified (it is possible and valid
222used, meaning the node will bind on a dynamically-assigned port on every 299used, meaning the node will bind on a dynamically-assigned port on every
223local IP address it finds. 300local IP address it finds.
224 301
225=item step 3, connect to seed nodes 302=item step 3, connect to seed nodes
226 303
227As the last step, the seeds list from the profile is passed to the 304As the last step, the seed ID list from the profile is passed to the
228L<AnyEvent::MP::Global> module, which will then use it to keep 305L<AnyEvent::MP::Global> module, which will then use it to keep
229connectivity with at least one node at any point in time. 306connectivity with at least one node at any point in time.
230 307
231=back 308=back
232 309
233Example: become a distributed node using the locla node name as profile. 310Example: become a distributed node using the local node name as profile.
234This should be the most common form of invocation for "daemon"-type nodes. 311This should be the most common form of invocation for "daemon"-type nodes.
235 312
236 configure 313 configure
237 314
238Example: become an anonymous node. This form is often used for commandline 315Example: become a semi-anonymous node. This form is often used for
239clients. 316commandline clients.
240 317
241 configure nodeid => "anon/"; 318 configure nodeid => "myscript/%n/%u";
242 319
243Example: configure a node using a profile called seed, which si suitable 320Example: configure a node using a profile called seed, which is suitable
244for a seed node as it binds on all local addresses on a fixed port (4040, 321for a seed node as it binds on all local addresses on a fixed port (4040,
245customary for aemp). 322customary for aemp).
246 323
247 # use the aemp commandline utility 324 # use the aemp commandline utility
248 # aemp profile seed nodeid anon/ binds '*:4040' 325 # aemp profile seed binds '*:4040'
249 326
250 # then use it 327 # then use it
251 configure profile => "seed"; 328 configure profile => "seed";
252 329
253 # or simply use aemp from the shell again: 330 # or simply use aemp from the shell again:
318 395
319=cut 396=cut
320 397
321sub rcv($@); 398sub rcv($@);
322 399
323sub _kilme { 400my $KILME = sub {
324 die "received message on port without callback"; 401 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
325} 402 kil $SELF, unhandled_message => "no callback found for message '$tag'";
403};
326 404
327sub port(;&) { 405sub port(;&) {
328 my $id = "$UNIQ." . $ID++; 406 my $id = $UNIQ . ++$ID;
329 my $port = "$NODE#$id"; 407 my $port = "$NODE#$id";
330 408
331 rcv $port, shift || \&_kilme; 409 rcv $port, shift || $KILME;
332 410
333 $port 411 $port
334} 412}
335 413
336=item rcv $local_port, $callback->(@msg) 414=item rcv $local_port, $callback->(@msg)
341 419
342The global C<$SELF> (exported by this module) contains C<$port> while 420The global C<$SELF> (exported by this module) contains C<$port> while
343executing the callback. Runtime errors during callback execution will 421executing the callback. Runtime errors during callback execution will
344result in the port being C<kil>ed. 422result in the port being C<kil>ed.
345 423
346The default callback received all messages not matched by a more specific 424The default callback receives all messages not matched by a more specific
347C<tag> match. 425C<tag> match.
348 426
349=item rcv $local_port, tag => $callback->(@msg_without_tag), ... 427=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
350 428
351Register (or replace) callbacks to be called on messages starting with the 429Register (or replace) callbacks to be called on messages starting with the
372 msg1 => sub { ... }, 450 msg1 => sub { ... },
373 ... 451 ...
374 ; 452 ;
375 453
376Example: temporarily register a rcv callback for a tag matching some port 454Example: temporarily register a rcv callback for a tag matching some port
377(e.g. for a rpc reply) and unregister it after a message was received. 455(e.g. for an rpc reply) and unregister it after a message was received.
378 456
379 rcv $port, $otherport => sub { 457 rcv $port, $otherport => sub {
380 my @reply = @_; 458 my @reply = @_;
381 459
382 rcv $SELF, $otherport; 460 rcv $SELF, $otherport;
395 if (ref $_[0]) { 473 if (ref $_[0]) {
396 if (my $self = $PORT_DATA{$portid}) { 474 if (my $self = $PORT_DATA{$portid}) {
397 "AnyEvent::MP::Port" eq ref $self 475 "AnyEvent::MP::Port" eq ref $self
398 or Carp::croak "$port: rcv can only be called on message matching ports, caught"; 476 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
399 477
400 $self->[2] = shift; 478 $self->[0] = shift;
401 } else { 479 } else {
402 my $cb = shift; 480 my $cb = shift;
403 $PORT{$portid} = sub { 481 $PORT{$portid} = sub {
404 local $SELF = $port; 482 local $SELF = $port;
405 eval { &$cb }; _self_die if $@; 483 eval { &$cb }; _self_die if $@;
406 }; 484 };
407 } 485 }
408 } elsif (defined $_[0]) { 486 } elsif (defined $_[0]) {
409 my $self = $PORT_DATA{$portid} ||= do { 487 my $self = $PORT_DATA{$portid} ||= do {
410 my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port"; 488 my $self = bless [$PORT{$portid} || sub { }, { }, $port], "AnyEvent::MP::Port";
411 489
412 $PORT{$portid} = sub { 490 $PORT{$portid} = sub {
413 local $SELF = $port; 491 local $SELF = $port;
414 492
415 if (my $cb = $self->[1]{$_[0]}) { 493 if (my $cb = $self->[1]{$_[0]}) {
437 } 515 }
438 516
439 $port 517 $port
440} 518}
441 519
520=item peval $port, $coderef[, @args]
521
522Evaluates the given C<$codref> within the contetx of C<$port>, that is,
523when the code throews an exception the C<$port> will be killed.
524
525Any remaining args will be passed to the callback. Any return values will
526be returned to the caller.
527
528This is useful when you temporarily want to execute code in the context of
529a port.
530
531Example: create a port and run some initialisation code in it's context.
532
533 my $port = port { ... };
534
535 peval $port, sub {
536 init
537 or die "unable to init";
538 };
539
540=cut
541
542sub peval($$) {
543 local $SELF = shift;
544 my $cb = shift;
545
546 if (wantarray) {
547 my @res = eval { &$cb };
548 _self_die if $@;
549 @res
550 } else {
551 my $res = eval { &$cb };
552 _self_die if $@;
553 $res
554 }
555}
556
442=item $closure = psub { BLOCK } 557=item $closure = psub { BLOCK }
443 558
444Remembers C<$SELF> and creates a closure out of the BLOCK. When the 559Remembers C<$SELF> and creates a closure out of the BLOCK. When the
445closure is executed, sets up the environment in the same way as in C<rcv> 560closure is executed, sets up the environment in the same way as in C<rcv>
446callbacks, i.e. runtime errors will cause the port to get C<kil>ed. 561callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
562
563The effect is basically as if it returned C<< sub { peval $SELF, sub {
564BLOCK }, @_ } >>.
447 565
448This is useful when you register callbacks from C<rcv> callbacks: 566This is useful when you register callbacks from C<rcv> callbacks:
449 567
450 rcv delayed_reply => sub { 568 rcv delayed_reply => sub {
451 my ($delay, @reply) = @_; 569 my ($delay, @reply) = @_;
475 $res 593 $res
476 } 594 }
477 } 595 }
478} 596}
479 597
598=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
599
600=item $guard = mon $port # kill $SELF when $port dies
601
480=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies 602=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
481
482=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
483
484=item $guard = mon $port # kill $SELF when $port dies
485 603
486=item $guard = mon $port, $rcvport, @msg # send a message when $port dies 604=item $guard = mon $port, $rcvport, @msg # send a message when $port dies
487 605
488Monitor the given port and do something when the port is killed or 606Monitor the given port and do something when the port is killed or
489messages to it were lost, and optionally return a guard that can be used 607messages to it were lost, and optionally return a guard that can be used
490to stop monitoring again. 608to stop monitoring again.
491 609
610The first two forms distinguish between "normal" and "abnormal" kil's:
611
612In the first form (another port given), if the C<$port> is C<kil>'ed with
613a non-empty reason, the other port (C<$rcvport>) will be kil'ed with the
614same reason. That is, on "normal" kil's nothing happens, while under all
615other conditions, the other port is killed with the same reason.
616
617The second form (kill self) is the same as the first form, except that
618C<$rvport> defaults to C<$SELF>.
619
620The remaining forms don't distinguish between "normal" and "abnormal" kil's
621- it's up to the callback or receiver to check whether the C<@reason> is
622empty and act accordingly.
623
492In the first form (callback), the callback is simply called with any 624In the third form (callback), the callback is simply called with any
493number of C<@reason> elements (no @reason means that the port was deleted 625number of C<@reason> elements (empty @reason means that the port was deleted
494"normally"). Note also that I<< the callback B<must> never die >>, so use 626"normally"). Note also that I<< the callback B<must> never die >>, so use
495C<eval> if unsure. 627C<eval> if unsure.
496 628
497In the second form (another port given), the other port (C<$rcvport>)
498will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
499"normal" kils nothing happens, while under all other conditions, the other
500port is killed with the same reason.
501
502The third form (kill self) is the same as the second form, except that
503C<$rvport> defaults to C<$SELF>.
504
505In the last form (message), a message of the form C<@msg, @reason> will be 629In the last form (message), a message of the form C<$rcvport, @msg,
506C<snd>. 630@reason> will be C<snd>.
507 631
508Monitoring-actions are one-shot: once messages are lost (and a monitoring 632Monitoring-actions are one-shot: once messages are lost (and a monitoring
509alert was raised), they are removed and will not trigger again. 633alert was raised), they are removed and will not trigger again, even if it
634turns out that the port is still alive.
510 635
511As a rule of thumb, monitoring requests should always monitor a port from 636As a rule of thumb, monitoring requests should always monitor a remote
512a local port (or callback). The reason is that kill messages might get 637port locally (using a local C<$rcvport> or a callback). The reason is that
513lost, just like any other message. Another less obvious reason is that 638kill messages might get lost, just like any other message. Another less
514even monitoring requests can get lost (for example, when the connection 639obvious reason is that even monitoring requests can get lost (for example,
515to the other node goes down permanently). When monitoring a port locally 640when the connection to the other node goes down permanently). When
516these problems do not exist. 641monitoring a port locally these problems do not exist.
517 642
518C<mon> effectively guarantees that, in the absence of hardware failures, 643C<mon> effectively guarantees that, in the absence of hardware failures,
519after starting the monitor, either all messages sent to the port will 644after starting the monitor, either all messages sent to the port will
520arrive, or the monitoring action will be invoked after possible message 645arrive, or the monitoring action will be invoked after possible message
521loss has been detected. No messages will be lost "in between" (after 646loss has been detected. No messages will be lost "in between" (after
524delivered again. 649delivered again.
525 650
526Inter-host-connection timeouts and monitoring depend on the transport 651Inter-host-connection timeouts and monitoring depend on the transport
527used. The only transport currently implemented is TCP, and AnyEvent::MP 652used. The only transport currently implemented is TCP, and AnyEvent::MP
528relies on TCP to detect node-downs (this can take 10-15 minutes on a 653relies on TCP to detect node-downs (this can take 10-15 minutes on a
529non-idle connection, and usually around two hours for idle conenctions). 654non-idle connection, and usually around two hours for idle connections).
530 655
531This means that monitoring is good for program errors and cleaning up 656This means that monitoring is good for program errors and cleaning up
532stuff eventually, but they are no replacement for a timeout when you need 657stuff eventually, but they are no replacement for a timeout when you need
533to ensure some maximum latency. 658to ensure some maximum latency.
534 659
566 } 691 }
567 692
568 $node->monitor ($port, $cb); 693 $node->monitor ($port, $cb);
569 694
570 defined wantarray 695 defined wantarray
571 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) } 696 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
572} 697}
573 698
574=item $guard = mon_guard $port, $ref, $ref... 699=item $guard = mon_guard $port, $ref, $ref...
575 700
576Monitors the given C<$port> and keeps the passed references. When the port 701Monitors the given C<$port> and keeps the passed references. When the port
599 724
600=item kil $port[, @reason] 725=item kil $port[, @reason]
601 726
602Kill the specified port with the given C<@reason>. 727Kill the specified port with the given C<@reason>.
603 728
604If no C<@reason> is specified, then the port is killed "normally" (ports 729If no C<@reason> is specified, then the port is killed "normally" -
605monitoring other ports will not necessarily die because a port dies 730monitor callback will be invoked, but the kil will not cause linked ports
606"normally"). 731(C<mon $mport, $lport> form) to get killed.
607 732
608Otherwise, linked ports get killed with the same reason (second form of 733If a C<@reason> is specified, then linked ports (C<mon $mport, $lport>
609C<mon>, see above). 734form) get killed with the same reason.
610 735
611Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 736Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
612will be reported as reason C<< die => $@ >>. 737will be reported as reason C<< die => $@ >>.
613 738
614Transport/communication errors are reported as C<< transport_error => 739Transport/communication errors are reported as C<< transport_error =>
615$message >>. 740$message >>.
616 741
617=cut 742Common idioms:
743
744 # silently remove yourself, do not kill linked ports
745 kil $SELF;
746
747 # report a failure in some detail
748 kil $SELF, failure_mode_1 => "it failed with too high temperature";
749
750 # do not waste much time with killing, just die when something goes wrong
751 open my $fh, "<file"
752 or die "file: $!";
618 753
619=item $port = spawn $node, $initfunc[, @initdata] 754=item $port = spawn $node, $initfunc[, @initdata]
620 755
621Creates a port on the node C<$node> (which can also be a port ID, in which 756Creates a port on the node C<$node> (which can also be a port ID, in which
622case it's the node where that port resides). 757case it's the node where that port resides).
680} 815}
681 816
682sub spawn(@) { 817sub spawn(@) {
683 my ($nodeid, undef) = split /#/, shift, 2; 818 my ($nodeid, undef) = split /#/, shift, 2;
684 819
685 my $id = "$RUNIQ." . $ID++; 820 my $id = $RUNIQ . ++$ID;
686 821
687 $_[0] =~ /::/ 822 $_[0] =~ /::/
688 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 823 or Carp::croak "spawn init function must be a fully-qualified name, caught";
689 824
690 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_; 825 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
691 826
692 "$nodeid#$id" 827 "$nodeid#$id"
693} 828}
829
694 830
695=item after $timeout, @msg 831=item after $timeout, @msg
696 832
697=item after $timeout, $callback 833=item after $timeout, $callback
698 834
714 ? $action[0]() 850 ? $action[0]()
715 : snd @action; 851 : snd @action;
716 }; 852 };
717} 853}
718 854
855#=item $cb2 = timeout $seconds, $cb[, @args]
856
857=item cal $port, @msg, $callback[, $timeout]
858
859A simple form of RPC - sends a message to the given C<$port> with the
860given contents (C<@msg>), but adds a reply port to the message.
861
862The reply port is created temporarily just for the purpose of receiving
863the reply, and will be C<kil>ed when no longer needed.
864
865A reply message sent to the port is passed to the C<$callback> as-is.
866
867If an optional time-out (in seconds) is given and it is not C<undef>,
868then the callback will be called without any arguments after the time-out
869elapsed and the port is C<kil>ed.
870
871If no time-out is given (or it is C<undef>), then the local port will
872monitor the remote port instead, so it eventually gets cleaned-up.
873
874Currently this function returns the temporary port, but this "feature"
875might go in future versions unless you can make a convincing case that
876this is indeed useful for something.
877
878=cut
879
880sub cal(@) {
881 my $timeout = ref $_[-1] ? undef : pop;
882 my $cb = pop;
883
884 my $port = port {
885 undef $timeout;
886 kil $SELF;
887 &$cb;
888 };
889
890 if (defined $timeout) {
891 $timeout = AE::timer $timeout, 0, sub {
892 undef $timeout;
893 kil $port;
894 $cb->();
895 };
896 } else {
897 mon $_[0], sub {
898 kil $port;
899 $cb->();
900 };
901 }
902
903 push @_, $port;
904 &snd;
905
906 $port
907}
908
909=back
910
911=head1 DISTRIBUTED DATABASE
912
913AnyEvent::MP comes with a simple distributed database. The database will
914be mirrored asynchronously on all global nodes. Other nodes bind to one
915of the global nodes for their needs. Every node has a "local database"
916which contains all the values that are set locally. All local databases
917are merged together to form the global database, which can be queried.
918
919The database structure is that of a two-level hash - the database hash
920contains hashes which contain values, similarly to a perl hash of hashes,
921i.e.:
922
923 $DATABASE{$family}{$subkey} = $value
924
925The top level hash key is called "family", and the second-level hash key
926is called "subkey" or simply "key".
927
928The family must be alphanumeric, i.e. start with a letter and consist
929of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>,
930pretty much like Perl module names.
931
932As the family namespace is global, it is recommended to prefix family names
933with the name of the application or module using it.
934
935The subkeys must be non-empty strings, with no further restrictions.
936
937The values should preferably be strings, but other perl scalars should
938work as well (such as C<undef>, arrays and hashes).
939
940Every database entry is owned by one node - adding the same family/subkey
941combination on multiple nodes will not cause discomfort for AnyEvent::MP,
942but the result might be nondeterministic, i.e. the key might have
943different values on different nodes.
944
945Different subkeys in the same family can be owned by different nodes
946without problems, and in fact, this is the common method to create worker
947pools. For example, a worker port for image scaling might do this:
948
949 db_set my_image_scalers => $port;
950
951And clients looking for an image scaler will want to get the
952C<my_image_scalers> keys from time to time:
953
954 db_keys my_image_scalers => sub {
955 @ports = @{ $_[0] };
956 };
957
958Or better yet, they want to monitor the database family, so they always
959have a reasonable up-to-date copy:
960
961 db_mon my_image_scalers => sub {
962 @ports = keys %{ $_[0] };
963 };
964
965In general, you can set or delete single subkeys, but query and monitor
966whole families only.
967
968If you feel the need to monitor or query a single subkey, try giving it
969it's own family.
970
971=over
972
973=item $guard = db_set $family => $subkey [=> $value]
974
975Sets (or replaces) a key to the database - if C<$value> is omitted,
976C<undef> is used instead.
977
978When called in non-void context, C<db_set> returns a guard that
979automatically calls C<db_del> when it is destroyed.
980
981=item db_del $family => $subkey...
982
983Deletes one or more subkeys from the database family.
984
985=item $guard = db_reg $family => $port => $value
986
987=item $guard = db_reg $family => $port
988
989=item $guard = db_reg $family
990
991Registers a port in the given family and optionally returns a guard to
992remove it.
993
994This function basically does the same as:
995
996 db_set $family => $port => $value
997
998Except that the port is monitored and automatically removed from the
999database family when it is kil'ed.
1000
1001If C<$value> is missing, C<undef> is used. If C<$port> is missing, then
1002C<$SELF> is used.
1003
1004This function is most useful to register a port in some port group (which
1005is just another name for a database family), and have it removed when the
1006port is gone. This works best when the port is a local port.
1007
1008=cut
1009
1010sub db_reg($$;$) {
1011 my $family = shift;
1012 my $port = @_ ? shift : $SELF;
1013
1014 my $clr = sub { db_del $family => $port };
1015 mon $port, $clr;
1016
1017 db_set $family => $port => $_[0];
1018
1019 defined wantarray
1020 and &Guard::guard ($clr)
1021}
1022
1023=item db_family $family => $cb->(\%familyhash)
1024
1025Queries the named database C<$family> and call the callback with the
1026family represented as a hash. You can keep and freely modify the hash.
1027
1028=item db_keys $family => $cb->(\@keys)
1029
1030Same as C<db_family>, except it only queries the family I<subkeys> and passes
1031them as array reference to the callback.
1032
1033=item db_values $family => $cb->(\@values)
1034
1035Same as C<db_family>, except it only queries the family I<values> and passes them
1036as array reference to the callback.
1037
1038=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
1039
1040Creates a monitor on the given database family. Each time a key is set
1041or or is deleted the callback is called with a hash containing the
1042database family and three lists of added, changed and deleted subkeys,
1043respectively. If no keys have changed then the array reference might be
1044C<undef> or even missing.
1045
1046If not called in void context, a guard object is returned that, when
1047destroyed, stops the monitor.
1048
1049The family hash reference and the key arrays belong to AnyEvent::MP and
1050B<must not be modified or stored> by the callback. When in doubt, make a
1051copy.
1052
1053As soon as possible after the monitoring starts, the callback will be
1054called with the intiial contents of the family, even if it is empty,
1055i.e. there will always be a timely call to the callback with the current
1056contents.
1057
1058It is possible that the callback is called with a change event even though
1059the subkey is already present and the value has not changed.
1060
1061The monitoring stops when the guard object is destroyed.
1062
1063Example: on every change to the family "mygroup", print out all keys.
1064
1065 my $guard = db_mon mygroup => sub {
1066 my ($family, $a, $c, $d) = @_;
1067 print "mygroup members: ", (join " ", keys %$family), "\n";
1068 };
1069
1070Exmaple: wait until the family "My::Module::workers" is non-empty.
1071
1072 my $guard; $guard = db_mon My::Module::workers => sub {
1073 my ($family, $a, $c, $d) = @_;
1074 return unless %$family;
1075 undef $guard;
1076 print "My::Module::workers now nonempty\n";
1077 };
1078
1079Example: print all changes to the family "AnyRvent::Fantasy::Module".
1080
1081 my $guard = db_mon AnyRvent::Fantasy::Module => sub {
1082 my ($family, $a, $c, $d) = @_;
1083
1084 print "+$_=$family->{$_}\n" for @$a;
1085 print "*$_=$family->{$_}\n" for @$c;
1086 print "-$_=$family->{$_}\n" for @$d;
1087 };
1088
1089=cut
1090
719=back 1091=back
720 1092
721=head1 AnyEvent::MP vs. Distributed Erlang 1093=head1 AnyEvent::MP vs. Distributed Erlang
722 1094
723AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 1095AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
724== aemp node, Erlang process == aemp port), so many of the documents and 1096== aemp node, Erlang process == aemp port), so many of the documents and
725programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 1097programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
726sample: 1098sample:
727 1099
728 http://www.Erlang.se/doc/programming_rules.shtml 1100 http://www.erlang.se/doc/programming_rules.shtml
729 http://Erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4 1101 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
730 http://Erlang.org/download/Erlang-book-part1.pdf # chapters 5 and 6 1102 http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6
731 http://Erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5 1103 http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
732 1104
733Despite the similarities, there are also some important differences: 1105Despite the similarities, there are also some important differences:
734 1106
735=over 4 1107=over 4
736 1108
737=item * Node IDs are arbitrary strings in AEMP. 1109=item * Node IDs are arbitrary strings in AEMP.
738 1110
739Erlang relies on special naming and DNS to work everywhere in the same 1111Erlang relies on special naming and DNS to work everywhere in the same
740way. AEMP relies on each node somehow knowing its own address(es) (e.g. by 1112way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
741configuration or DNS), but will otherwise discover other odes itself. 1113configuration or DNS), and possibly the addresses of some seed nodes, but
1114will otherwise discover other nodes (and their IDs) itself.
742 1115
743=item * Erlang has a "remote ports are like local ports" philosophy, AEMP 1116=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
744uses "local ports are like remote ports". 1117uses "local ports are like remote ports".
745 1118
746The failure modes for local ports are quite different (runtime errors 1119The failure modes for local ports are quite different (runtime errors
755ports being the special case/exception, where transport errors cannot 1128ports being the special case/exception, where transport errors cannot
756occur. 1129occur.
757 1130
758=item * Erlang uses processes and a mailbox, AEMP does not queue. 1131=item * Erlang uses processes and a mailbox, AEMP does not queue.
759 1132
760Erlang uses processes that selectively receive messages, and therefore 1133Erlang uses processes that selectively receive messages out of order, and
761needs a queue. AEMP is event based, queuing messages would serve no 1134therefore needs a queue. AEMP is event based, queuing messages would serve
762useful purpose. For the same reason the pattern-matching abilities of 1135no useful purpose. For the same reason the pattern-matching abilities
763AnyEvent::MP are more limited, as there is little need to be able to 1136of AnyEvent::MP are more limited, as there is little need to be able to
764filter messages without dequeuing them. 1137filter messages without dequeuing them.
765 1138
766(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 1139This is not a philosophical difference, but simply stems from AnyEvent::MP
1140being event-based, while Erlang is process-based.
1141
1142You cna have a look at L<Coro::MP> for a more Erlang-like process model on
1143top of AEMP and Coro threads.
767 1144
768=item * Erlang sends are synchronous, AEMP sends are asynchronous. 1145=item * Erlang sends are synchronous, AEMP sends are asynchronous.
769 1146
770Sending messages in Erlang is synchronous and blocks the process (and 1147Sending messages in Erlang is synchronous and blocks the process until
1148a conenction has been established and the message sent (and so does not
771so does not need a queue that can overflow). AEMP sends are immediate, 1149need a queue that can overflow). AEMP sends return immediately, connection
772connection establishment is handled in the background. 1150establishment is handled in the background.
773 1151
774=item * Erlang suffers from silent message loss, AEMP does not. 1152=item * Erlang suffers from silent message loss, AEMP does not.
775 1153
776Erlang makes few guarantees on messages delivery - messages can get lost 1154Erlang implements few guarantees on messages delivery - messages can get
777without any of the processes realising it (i.e. you send messages a, b, 1155lost without any of the processes realising it (i.e. you send messages a,
778and c, and the other side only receives messages a and c). 1156b, and c, and the other side only receives messages a and c).
779 1157
780AEMP guarantees correct ordering, and the guarantee that after one message 1158AEMP guarantees (modulo hardware errors) correct ordering, and the
781is lost, all following ones sent to the same port are lost as well, until 1159guarantee that after one message is lost, all following ones sent to the
782monitoring raises an error, so there are no silent "holes" in the message 1160same port are lost as well, until monitoring raises an error, so there are
783sequence. 1161no silent "holes" in the message sequence.
1162
1163If you want your software to be very reliable, you have to cope with
1164corrupted and even out-of-order messages in both Erlang and AEMP. AEMP
1165simply tries to work better in common error cases, such as when a network
1166link goes down.
784 1167
785=item * Erlang can send messages to the wrong port, AEMP does not. 1168=item * Erlang can send messages to the wrong port, AEMP does not.
786 1169
787In Erlang it is quite likely that a node that restarts reuses a process ID 1170In Erlang it is quite likely that a node that restarts reuses an Erlang
788known to other nodes for a completely different process, causing messages 1171process ID known to other nodes for a completely different process,
789destined for that process to end up in an unrelated process. 1172causing messages destined for that process to end up in an unrelated
1173process.
790 1174
791AEMP never reuses port IDs, so old messages or old port IDs floating 1175AEMP does not reuse port IDs, so old messages or old port IDs floating
792around in the network will not be sent to an unrelated port. 1176around in the network will not be sent to an unrelated port.
793 1177
794=item * Erlang uses unprotected connections, AEMP uses secure 1178=item * Erlang uses unprotected connections, AEMP uses secure
795authentication and can use TLS. 1179authentication and can use TLS.
796 1180
799 1183
800=item * The AEMP protocol is optimised for both text-based and binary 1184=item * The AEMP protocol is optimised for both text-based and binary
801communications. 1185communications.
802 1186
803The AEMP protocol, unlike the Erlang protocol, supports both programming 1187The AEMP protocol, unlike the Erlang protocol, supports both programming
804language independent text-only protocols (good for debugging) and binary, 1188language independent text-only protocols (good for debugging), and binary,
805language-specific serialisers (e.g. Storable). By default, unless TLS is 1189language-specific serialisers (e.g. Storable). By default, unless TLS is
806used, the protocol is actually completely text-based. 1190used, the protocol is actually completely text-based.
807 1191
808It has also been carefully designed to be implementable in other languages 1192It has also been carefully designed to be implementable in other languages
809with a minimum of work while gracefully degrading functionality to make the 1193with a minimum of work while gracefully degrading functionality to make the
810protocol simple. 1194protocol simple.
811 1195
812=item * AEMP has more flexible monitoring options than Erlang. 1196=item * AEMP has more flexible monitoring options than Erlang.
813 1197
814In Erlang, you can chose to receive I<all> exit signals as messages 1198In Erlang, you can chose to receive I<all> exit signals as messages or
815or I<none>, there is no in-between, so monitoring single processes is 1199I<none>, there is no in-between, so monitoring single Erlang processes is
816difficult to implement. Monitoring in AEMP is more flexible than in 1200difficult to implement.
817Erlang, as one can choose between automatic kill, exit message or callback 1201
818on a per-process basis. 1202Monitoring in AEMP is more flexible than in Erlang, as one can choose
1203between automatic kill, exit message or callback on a per-port basis.
819 1204
820=item * Erlang tries to hide remote/local connections, AEMP does not. 1205=item * Erlang tries to hide remote/local connections, AEMP does not.
821 1206
822Monitoring in Erlang is not an indicator of process death/crashes, in the 1207Monitoring in Erlang is not an indicator of process death/crashes, in the
823same way as linking is (except linking is unreliable in Erlang). 1208same way as linking is (except linking is unreliable in Erlang).
845overhead, as well as having to keep a proxy object everywhere. 1230overhead, as well as having to keep a proxy object everywhere.
846 1231
847Strings can easily be printed, easily serialised etc. and need no special 1232Strings can easily be printed, easily serialised etc. and need no special
848procedures to be "valid". 1233procedures to be "valid".
849 1234
850And as a result, a miniport consists of a single closure stored in a 1235And as a result, a port with just a default receiver consists of a single
851global hash - it can't become much cheaper. 1236code reference stored in a global hash - it can't become much cheaper.
852 1237
853=item Why favour JSON, why not a real serialising format such as Storable? 1238=item Why favour JSON, why not a real serialising format such as Storable?
854 1239
855In fact, any AnyEvent::MP node will happily accept Storable as framing 1240In fact, any AnyEvent::MP node will happily accept Storable as framing
856format, but currently there is no way to make a node use Storable by 1241format, but currently there is no way to make a node use Storable by
866Keeping your messages simple, concentrating on data structures rather than 1251Keeping your messages simple, concentrating on data structures rather than
867objects, will keep your messages clean, tidy and efficient. 1252objects, will keep your messages clean, tidy and efficient.
868 1253
869=back 1254=back
870 1255
1256=head1 PORTING FROM AnyEvent::MP VERSION 1.X
1257
1258AEMP version 2 has a few major incompatible changes compared to version 1:
1259
1260=over 4
1261
1262=item AnyEvent::MP::Global no longer has group management functions.
1263
1264At least not officially - the grp_* functions are still exported and might
1265work, but they will be removed in some later release.
1266
1267AnyEvent::MP now comes with a distributed database that is more
1268powerful. Its database families map closely to port groups, but the API
1269has changed (the functions are also now exported by AnyEvent::MP). Here is
1270a rough porting guide:
1271
1272 grp_reg $group, $port # old
1273 db_reg $group, $port # new
1274
1275 $list = grp_get $group # old
1276 db_keys $group, sub { my $list = shift } # new
1277
1278 grp_mon $group, $cb->(\@ports, $add, $del) # old
1279 db_mon $group, $cb->(\%ports, $add, $change, $del) # new
1280
1281C<grp_reg> is a no-brainer (just replace by C<db_reg>), but C<grp_get> is
1282no longer instant, because the local node might not have a copy of the
1283group. You can either modify your code to allow for a callback, or use
1284C<db_mon> to keep an updated copy of the group:
1285
1286 my $local_group_copy;
1287 db_mon $group => sub { $local_group_copy = $_[0] };
1288
1289 # now "keys %$local_group_copy" always returns the most up-to-date
1290 # list of ports in the group.
1291
1292C<grp_mon> can be replaced by C<db_mon> with minor changes - C<db_mon>
1293passes a hash as first argument, and an extra C<$chg> argument that can be
1294ignored:
1295
1296 db_mon $group => sub {
1297 my ($ports, $add, $chg, $lde) = @_;
1298 $ports = [keys %$ports];
1299
1300 # now $ports, $add and $del are the same as
1301 # were originally passed by grp_mon.
1302 ...
1303 };
1304
1305=item Nodes not longer connect to all other nodes.
1306
1307In AEMP 1.x, every node automatically loads the L<AnyEvent::MP::Global>
1308module, which in turn would create connections to all other nodes in the
1309network (helped by the seed nodes).
1310
1311In version 2.x, global nodes still connect to all other global nodes, but
1312other nodes don't - now every node either is a global node itself, or
1313attaches itself to another global node.
1314
1315If a node isn't a global node itself, then it attaches itself to one
1316of its seed nodes. If that seed node isn't a global node yet, it will
1317automatically be upgraded to a global node.
1318
1319So in many cases, nothing needs to be changed - one just has to make sure
1320that all seed nodes are meshed together with the other seed nodes (as with
1321AEMP 1.x), and other nodes specify them as seed nodes. This is most easily
1322achieved by specifying the same set of seed nodes for all nodes in the
1323network.
1324
1325Not opening a connection to every other node is usually an advantage,
1326except when you need the lower latency of an already established
1327connection. To ensure a node establishes a connection to another node,
1328you can monitor the node port (C<mon $node, ...>), which will attempt to
1329create the connection (and notify you when the connection fails).
1330
1331=item Listener-less nodes (nodes without binds) are gone.
1332
1333And are not coming back, at least not in their old form. If no C<binds>
1334are specified for a node, AnyEvent::MP assumes a default of C<*:*>.
1335
1336There are vague plans to implement some form of routing domains, which
1337might or might not bring back listener-less nodes, but don't count on it.
1338
1339The fact that most connections are now optional somewhat mitigates this,
1340as a node can be effectively unreachable from the outside without any
1341problems, as long as it isn't a global node and only reaches out to other
1342nodes (as opposed to being contacted from other nodes).
1343
1344=item $AnyEvent::MP::Kernel::WARN has gone.
1345
1346AnyEvent has acquired a logging framework (L<AnyEvent::Log>), and AEMP now
1347uses this, and so should your programs.
1348
1349Every module now documents what kinds of messages it generates, with
1350AnyEvent::MP acting as a catch all.
1351
1352On the positive side, this means that instead of setting
1353C<PERL_ANYEVENT_MP_WARNLEVEL>, you can get away by setting C<AE_VERBOSE> -
1354much less to type.
1355
1356=back
1357
1358=head1 LOGGING
1359
1360AnyEvent::MP does not normally log anything by itself, but sinc eit is the
1361root of the contetx hierarchy for AnyEvent::MP modules, it will receive
1362all log messages by submodules.
1363
871=head1 SEE ALSO 1364=head1 SEE ALSO
872 1365
873L<AnyEvent::MP::Intro> - a gentle introduction. 1366L<AnyEvent::MP::Intro> - a gentle introduction.
874 1367
875L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 1368L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
876 1369
877L<AnyEvent::MP::Global> - network maintainance and port groups, to find 1370L<AnyEvent::MP::Global> - network maintenance and port groups, to find
878your applications. 1371your applications.
1372
1373L<AnyEvent::MP::DataConn> - establish data connections between nodes.
879 1374
880L<AnyEvent::MP::LogCatcher> - simple service to display log messages from 1375L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
881all nodes. 1376all nodes.
882 1377
883L<AnyEvent>. 1378L<AnyEvent>.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines