ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-MP/MP.pm
Revision: 1.124
Committed: Sat Mar 3 11:38:43 2012 UTC (12 years, 2 months ago) by root
Branch: MAIN
Changes since 1.123: +48 -1 lines
Log Message:
*** empty log message ***

File Contents

# User Rev Content
1 root 1.1 =head1 NAME
2    
3 root 1.89 AnyEvent::MP - erlang-style multi-processing/message-passing framework
4 root 1.1
5     =head1 SYNOPSIS
6    
7     use AnyEvent::MP;
8    
9 root 1.75 $NODE # contains this node's node ID
10     NODE # returns this node's node ID
11 root 1.2
12 root 1.38 $SELF # receiving/own port id in rcv callbacks
13    
14 root 1.48 # initialise the node so it can send/receive messages
15 root 1.72 configure;
16 root 1.48
17 root 1.75 # ports are message destinations
18 root 1.38
19     # sending messages
20 root 1.2 snd $port, type => data...;
21 root 1.38 snd $port, @msg;
22     snd @msg_with_first_element_being_a_port;
23 root 1.2
24 root 1.50 # creating/using ports, the simple way
25 root 1.73 my $simple_port = port { my @msg = @_ };
26 root 1.22
27 root 1.52 # creating/using ports, tagged message matching
28 root 1.38 my $port = port;
29 root 1.73 rcv $port, ping => sub { snd $_[0], "pong" };
30     rcv $port, pong => sub { warn "pong received\n" };
31 root 1.2
32 root 1.48 # create a port on another node
33     my $port = spawn $node, $initfunc, @initdata;
34    
35 root 1.116 # destroy a port again
36 root 1.101 kil $port; # "normal" kill
37     kil $port, my_error => "everything is broken"; # error kill
38    
39 root 1.35 # monitoring
40 root 1.90 mon $localport, $cb->(@msg) # callback is invoked on death
41     mon $localport, $otherport # kill otherport on abnormal death
42     mon $localport, $otherport, @msg # send message on death
43 root 1.35
44 root 1.101 # 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     };
51    
52 root 1.45 =head1 CURRENT STATUS
53    
54 root 1.71 bin/aemp - stable.
55     AnyEvent::MP - stable API, should work.
56 elmex 1.77 AnyEvent::MP::Intro - explains most concepts.
57 root 1.88 AnyEvent::MP::Kernel - mostly stable API.
58     AnyEvent::MP::Global - stable API.
59 root 1.45
60 root 1.1 =head1 DESCRIPTION
61    
62 root 1.2 This module (-family) implements a simple message passing framework.
63    
64     Despite its simplicity, you can securely message other processes running
65 root 1.67 on the same or other hosts, and you can supervise entities remotely.
66 root 1.2
67 root 1.23 For an introduction to this module family, see the L<AnyEvent::MP::Intro>
68 root 1.67 manual page and the examples under F<eg/>.
69 root 1.23
70 root 1.2 =head1 CONCEPTS
71    
72     =over 4
73    
74     =item port
75    
76 root 1.79 Not to be confused with a TCP port, a "port" is something you can send
77     messages to (with the C<snd> function).
78 root 1.29
79 root 1.53 Ports allow you to register C<rcv> handlers that can match all or just
80 root 1.64 some messages. Messages send to ports will not be queued, regardless of
81     anything was listening for them or not.
82 root 1.2
83 root 1.119 Ports are represented by (printable) strings called "port IDs".
84    
85 root 1.67 =item port ID - C<nodeid#portname>
86 root 1.2
87 root 1.123 A port ID is the concatenation of a node ID, a hash-mark (C<#>)
88     as separator, and a port name (a printable string of unspecified
89     format created by AnyEvent::MP).
90 root 1.2
91     =item node
92    
93 root 1.53 A node is a single process containing at least one port - the node port,
94 root 1.67 which enables nodes to manage each other remotely, and to create new
95 root 1.53 ports.
96 root 1.2
97 root 1.67 Nodes are either public (have one or more listening ports) or private
98     (no listening ports). Private nodes cannot talk to other private nodes
99 root 1.119 currently, but all nodes can talk to public nodes.
100    
101     Nodes is represented by (printable) strings called "node IDs".
102 root 1.2
103 root 1.117 =item node ID - C<[A-Za-z0-9_\-.:]*>
104 root 1.2
105 root 1.64 A node ID is a string that uniquely identifies the node within a
106     network. Depending on the configuration used, node IDs can look like a
107     hostname, a hostname and a port, or a random string. AnyEvent::MP itself
108 root 1.119 doesn't interpret node IDs in any way except to uniquely identify a node.
109 root 1.64
110     =item binds - C<ip:port>
111    
112     Nodes can only talk to each other by creating some kind of connection to
113     each other. To do this, nodes should listen on one or more local transport
114 root 1.119 endpoints - binds.
115    
116     Currently, only standard C<ip:port> specifications can be used, which
117     specify TCP ports to listen on. So a bind is basically just a tcp socket
118     in listening mode thta accepts conenctions form other nodes.
119 root 1.64
120 root 1.83 =item seed nodes
121 root 1.64
122 root 1.119 When a node starts, it knows nothing about the network it is in - it
123     needs to connect to at least one other node that is already in the
124     network. These other nodes are called "seed nodes".
125    
126     Seed nodes themselves are not special - they are seed nodes only because
127     some other node I<uses> them as such, but any node can be used as seed
128     node for other nodes, and eahc node cna use a different set of seed nodes.
129 root 1.83
130     In addition to discovering the network, seed nodes are also used to
131 root 1.119 maintain the network - all nodes using the same seed node form are part of
132     the same network. If a network is split into multiple subnets because e.g.
133     the network link between the parts goes down, then using the same seed
134     nodes for all nodes ensures that eventually the subnets get merged again.
135 root 1.83
136     Seed nodes are expected to be long-running, and at least one seed node
137 root 1.85 should always be available. They should also be relatively responsive - a
138     seed node that blocks for long periods will slow down everybody else.
139 root 1.83
140 root 1.119 For small networks, it's best if every node uses the same set of seed
141     nodes. For large networks, it can be useful to specify "regional" seed
142     nodes for most nodes in an area, and use all seed nodes as seed nodes for
143     each other. What's important is that all seed nodes connections form a
144     complete graph, so that the network cannot split into separate subnets
145     forever.
146    
147     Seed nodes are represented by seed IDs.
148    
149     =item seed IDs - C<host:port>
150 root 1.83
151 root 1.119 Seed IDs are transport endpoint(s) (usually a hostname/IP address and a
152 elmex 1.96 TCP port) of nodes that should be used as seed nodes.
153 root 1.29
154 root 1.119 =item global nodes
155    
156     An AEMP network needs a discovery service - nodes need to know how to
157     connect to other nodes they only know by name. In addition, AEMP offers a
158     distributed "group database", which maps group names to a list of strings
159     - for example, to register worker ports.
160    
161     A network needs at least one global node to work, and allows every node to
162     be a global node.
163    
164     Any node that loads the L<AnyEvent::MP::Global> module becomes a global
165     node and tries to keep connections to all other nodes. So while it can
166     make sense to make every node "global" in small networks, it usually makes
167     sense to only make seed nodes into global nodes in large networks (nodes
168     keep connections to seed nodes and global nodes, so makign them the same
169     reduces overhead).
170 root 1.67
171 root 1.2 =back
172    
173 root 1.3 =head1 VARIABLES/FUNCTIONS
174 root 1.2
175     =over 4
176    
177 root 1.1 =cut
178    
179     package AnyEvent::MP;
180    
181 root 1.121 use AnyEvent::MP::Config ();
182 root 1.44 use AnyEvent::MP::Kernel;
183 root 1.121 use AnyEvent::MP::Kernel qw(%NODE %PORT %PORT_DATA $UNIQ $RUNIQ $ID);
184 root 1.2
185 root 1.1 use common::sense;
186    
187 root 1.2 use Carp ();
188    
189 root 1.1 use AE ();
190 root 1.124 use Guard ();
191 root 1.1
192 root 1.2 use base "Exporter";
193    
194 root 1.121 our $VERSION = $AnyEvent::MP::Config::VERSION;
195 root 1.43
196 root 1.8 our @EXPORT = qw(
197 root 1.59 NODE $NODE *SELF node_of after
198 root 1.72 configure
199 root 1.101 snd rcv mon mon_guard kil psub peval spawn cal
200 root 1.22 port
201 root 1.124 db_set db_del db_reg
202 root 1.8 );
203 root 1.2
204 root 1.22 our $SELF;
205    
206     sub _self_die() {
207     my $msg = $@;
208     $msg =~ s/\n+$// unless ref $msg;
209     kil $SELF, die => $msg;
210     }
211    
212     =item $thisnode = NODE / $NODE
213    
214 root 1.67 The C<NODE> function returns, and the C<$NODE> variable contains, the node
215 root 1.64 ID of the node running in the current process. This value is initialised by
216 root 1.72 a call to C<configure>.
217 root 1.22
218 root 1.63 =item $nodeid = node_of $port
219 root 1.22
220 root 1.67 Extracts and returns the node ID from a port ID or a node ID.
221 root 1.34
222 root 1.78 =item configure $profile, key => value...
223    
224 root 1.72 =item configure key => value...
225 root 1.34
226 root 1.64 Before a node can talk to other nodes on the network (i.e. enter
227 root 1.72 "distributed mode") it has to configure itself - the minimum a node needs
228 root 1.64 to know is its own name, and optionally it should know the addresses of
229     some other nodes in the network to discover other nodes.
230 root 1.34
231 root 1.121 This function configures a node - it must be called exactly once (or
232     never) before calling other AnyEvent::MP functions.
233    
234 root 1.108 The key/value pairs are basically the same ones as documented for the
235 root 1.121 F<aemp> command line utility (sans the set/del prefix), with two additions:
236    
237     =over 4
238    
239     =item norc => $boolean (default false)
240    
241     If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not>
242     be consulted - all configuraiton options must be specified in the
243     C<configure> call.
244 root 1.108
245 root 1.121 =item force => $boolean (default false)
246    
247     IF true, then the values specified in the C<configure> will take
248     precedence over any values configured via the rc file. The default is for
249     the rc file to override any options specified in the program.
250    
251     =back
252 root 1.34
253 root 1.72 =over 4
254    
255     =item step 1, gathering configuration from profiles
256    
257     The function first looks up a profile in the aemp configuration (see the
258     L<aemp> commandline utility). The profile name can be specified via the
259 root 1.78 named C<profile> parameter or can simply be the first parameter). If it is
260     missing, then the nodename (F<uname -n>) will be used as profile name.
261 root 1.34
262 root 1.72 The profile data is then gathered as follows:
263 root 1.69
264 elmex 1.77 First, all remaining key => value pairs (all of which are conveniently
265 root 1.72 undocumented at the moment) will be interpreted as configuration
266     data. Then they will be overwritten by any values specified in the global
267     default configuration (see the F<aemp> utility), then the chain of
268     profiles chosen by the profile name (and any C<parent> attributes).
269    
270     That means that the values specified in the profile have highest priority
271     and the values specified directly via C<configure> have lowest priority,
272     and can only be used to specify defaults.
273 root 1.49
274 root 1.64 If the profile specifies a node ID, then this will become the node ID of
275 root 1.122 this process. If not, then the profile name will be used as node ID, with
276     a slash (C</>) attached.
277    
278     If the node ID (or profile name) ends with a slash (C</>), then a random
279     string is appended to make it unique.
280 root 1.64
281 root 1.72 =item step 2, bind listener sockets
282    
283 root 1.64 The next step is to look up the binds in the profile, followed by binding
284     aemp protocol listeners on all binds specified (it is possible and valid
285     to have no binds, meaning that the node cannot be contacted form the
286     outside. This means the node cannot talk to other nodes that also have no
287     binds, but it can still talk to all "normal" nodes).
288    
289 root 1.70 If the profile does not specify a binds list, then a default of C<*> is
290 root 1.72 used, meaning the node will bind on a dynamically-assigned port on every
291     local IP address it finds.
292    
293     =item step 3, connect to seed nodes
294 root 1.64
295 root 1.119 As the last step, the seed ID list from the profile is passed to the
296 root 1.64 L<AnyEvent::MP::Global> module, which will then use it to keep
297 root 1.72 connectivity with at least one node at any point in time.
298 root 1.64
299 root 1.72 =back
300    
301 root 1.87 Example: become a distributed node using the local node name as profile.
302 root 1.72 This should be the most common form of invocation for "daemon"-type nodes.
303 root 1.34
304 root 1.72 configure
305 root 1.34
306 root 1.64 Example: become an anonymous node. This form is often used for commandline
307     clients.
308 root 1.34
309 root 1.72 configure nodeid => "anon/";
310    
311 root 1.120 Example: configure a node using a profile called seed, which is suitable
312 root 1.72 for a seed node as it binds on all local addresses on a fixed port (4040,
313     customary for aemp).
314    
315     # use the aemp commandline utility
316 root 1.122 # aemp profile seed binds '*:4040'
317 root 1.72
318     # then use it
319     configure profile => "seed";
320 root 1.34
321 root 1.72 # or simply use aemp from the shell again:
322     # aemp run profile seed
323 root 1.34
324 root 1.72 # or provide a nicer-to-remember nodeid
325     # aemp run profile seed nodeid "$(hostname)"
326 root 1.34
327 root 1.22 =item $SELF
328    
329     Contains the current port id while executing C<rcv> callbacks or C<psub>
330     blocks.
331 root 1.3
332 root 1.67 =item *SELF, SELF, %SELF, @SELF...
333 root 1.22
334     Due to some quirks in how perl exports variables, it is impossible to
335 root 1.67 just export C<$SELF>, all the symbols named C<SELF> are exported by this
336 root 1.22 module, but only C<$SELF> is currently used.
337 root 1.3
338 root 1.33 =item snd $port, type => @data
339 root 1.3
340 root 1.33 =item snd $port, @msg
341 root 1.3
342 root 1.67 Send the given message to the given port, which can identify either a
343     local or a remote port, and must be a port ID.
344 root 1.8
345 root 1.67 While the message can be almost anything, it is highly recommended to
346     use a string as first element (a port ID, or some word that indicates a
347     request type etc.) and to consist if only simple perl values (scalars,
348     arrays, hashes) - if you think you need to pass an object, think again.
349    
350     The message data logically becomes read-only after a call to this
351     function: modifying any argument (or values referenced by them) is
352     forbidden, as there can be considerable time between the call to C<snd>
353     and the time the message is actually being serialised - in fact, it might
354     never be copied as within the same process it is simply handed to the
355     receiving port.
356 root 1.3
357     The type of data you can transfer depends on the transport protocol: when
358     JSON is used, then only strings, numbers and arrays and hashes consisting
359     of those are allowed (no objects). When Storable is used, then anything
360     that Storable can serialise and deserialise is allowed, and for the local
361 root 1.67 node, anything can be passed. Best rely only on the common denominator of
362     these.
363 root 1.3
364 root 1.22 =item $local_port = port
365 root 1.2
366 root 1.50 Create a new local port object and returns its port ID. Initially it has
367     no callbacks set and will throw an error when it receives messages.
368 root 1.10
369 root 1.50 =item $local_port = port { my @msg = @_ }
370 root 1.15
371 root 1.50 Creates a new local port, and returns its ID. Semantically the same as
372     creating a port and calling C<rcv $port, $callback> on it.
373 root 1.15
374 root 1.50 The block will be called for every message received on the port, with the
375     global variable C<$SELF> set to the port ID. Runtime errors will cause the
376     port to be C<kil>ed. The message will be passed as-is, no extra argument
377     (i.e. no port ID) will be passed to the callback.
378 root 1.15
379 root 1.50 If you want to stop/destroy the port, simply C<kil> it:
380 root 1.15
381 root 1.50 my $port = port {
382     my @msg = @_;
383     ...
384     kil $SELF;
385 root 1.15 };
386 root 1.10
387     =cut
388    
389 root 1.33 sub rcv($@);
390    
391 root 1.50 sub _kilme {
392     die "received message on port without callback";
393     }
394    
395 root 1.22 sub port(;&) {
396 root 1.123 my $id = $UNIQ . ++$ID;
397 root 1.22 my $port = "$NODE#$id";
398    
399 root 1.50 rcv $port, shift || \&_kilme;
400 root 1.10
401 root 1.22 $port
402 root 1.10 }
403    
404 root 1.50 =item rcv $local_port, $callback->(@msg)
405 root 1.31
406 root 1.50 Replaces the default callback on the specified port. There is no way to
407     remove the default callback: use C<sub { }> to disable it, or better
408     C<kil> the port when it is no longer needed.
409 root 1.3
410 root 1.33 The global C<$SELF> (exported by this module) contains C<$port> while
411 root 1.50 executing the callback. Runtime errors during callback execution will
412     result in the port being C<kil>ed.
413 root 1.22
414 root 1.50 The default callback received all messages not matched by a more specific
415     C<tag> match.
416 root 1.22
417 root 1.50 =item rcv $local_port, tag => $callback->(@msg_without_tag), ...
418 root 1.3
419 root 1.54 Register (or replace) callbacks to be called on messages starting with the
420     given tag on the given port (and return the port), or unregister it (when
421     C<$callback> is C<$undef> or missing). There can only be one callback
422     registered for each tag.
423 root 1.3
424 root 1.50 The original message will be passed to the callback, after the first
425     element (the tag) has been removed. The callback will use the same
426     environment as the default callback (see above).
427 root 1.3
428 root 1.36 Example: create a port and bind receivers on it in one go.
429    
430     my $port = rcv port,
431 root 1.50 msg1 => sub { ... },
432     msg2 => sub { ... },
433 root 1.36 ;
434    
435     Example: create a port, bind receivers and send it in a message elsewhere
436     in one go:
437    
438     snd $otherport, reply =>
439     rcv port,
440 root 1.50 msg1 => sub { ... },
441 root 1.36 ...
442     ;
443    
444 root 1.54 Example: temporarily register a rcv callback for a tag matching some port
445 root 1.102 (e.g. for an rpc reply) and unregister it after a message was received.
446 root 1.54
447     rcv $port, $otherport => sub {
448     my @reply = @_;
449    
450     rcv $SELF, $otherport;
451     };
452    
453 root 1.3 =cut
454    
455     sub rcv($@) {
456 root 1.33 my $port = shift;
457 root 1.75 my ($nodeid, $portid) = split /#/, $port, 2;
458 root 1.3
459 root 1.75 $NODE{$nodeid} == $NODE{""}
460 root 1.33 or Carp::croak "$port: rcv can only be called on local ports, caught";
461 root 1.22
462 root 1.50 while (@_) {
463     if (ref $_[0]) {
464     if (my $self = $PORT_DATA{$portid}) {
465     "AnyEvent::MP::Port" eq ref $self
466     or Carp::croak "$port: rcv can only be called on message matching ports, caught";
467 root 1.33
468 root 1.103 $self->[0] = shift;
469 root 1.50 } else {
470     my $cb = shift;
471     $PORT{$portid} = sub {
472     local $SELF = $port;
473     eval { &$cb }; _self_die if $@;
474     };
475     }
476     } elsif (defined $_[0]) {
477     my $self = $PORT_DATA{$portid} ||= do {
478 root 1.103 my $self = bless [$PORT{$portid} || sub { }, { }, $port], "AnyEvent::MP::Port";
479 root 1.50
480     $PORT{$portid} = sub {
481     local $SELF = $port;
482    
483     if (my $cb = $self->[1]{$_[0]}) {
484     shift;
485     eval { &$cb }; _self_die if $@;
486     } else {
487     &{ $self->[0] };
488 root 1.33 }
489     };
490 root 1.50
491     $self
492 root 1.33 };
493    
494 root 1.50 "AnyEvent::MP::Port" eq ref $self
495     or Carp::croak "$port: rcv can only be called on message matching ports, caught";
496 root 1.22
497 root 1.50 my ($tag, $cb) = splice @_, 0, 2;
498 root 1.33
499 root 1.50 if (defined $cb) {
500     $self->[1]{$tag} = $cb;
501 root 1.33 } else {
502 root 1.50 delete $self->[1]{$tag};
503 root 1.33 }
504 root 1.22 }
505 root 1.3 }
506 root 1.31
507 root 1.33 $port
508 root 1.2 }
509    
510 root 1.101 =item peval $port, $coderef[, @args]
511    
512     Evaluates the given C<$codref> within the contetx of C<$port>, that is,
513     when the code throews an exception the C<$port> will be killed.
514    
515     Any remaining args will be passed to the callback. Any return values will
516     be returned to the caller.
517    
518     This is useful when you temporarily want to execute code in the context of
519     a port.
520    
521     Example: create a port and run some initialisation code in it's context.
522    
523     my $port = port { ... };
524    
525     peval $port, sub {
526     init
527     or die "unable to init";
528     };
529    
530     =cut
531    
532     sub peval($$) {
533     local $SELF = shift;
534     my $cb = shift;
535    
536     if (wantarray) {
537     my @res = eval { &$cb };
538     _self_die if $@;
539     @res
540     } else {
541     my $res = eval { &$cb };
542     _self_die if $@;
543     $res
544     }
545     }
546    
547 root 1.22 =item $closure = psub { BLOCK }
548 root 1.2
549 root 1.22 Remembers C<$SELF> and creates a closure out of the BLOCK. When the
550     closure is executed, sets up the environment in the same way as in C<rcv>
551     callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
552    
553 root 1.101 The effect is basically as if it returned C<< sub { peval $SELF, sub {
554 root 1.114 BLOCK }, @_ } >>.
555 root 1.101
556 root 1.22 This is useful when you register callbacks from C<rcv> callbacks:
557    
558     rcv delayed_reply => sub {
559     my ($delay, @reply) = @_;
560     my $timer = AE::timer $delay, 0, psub {
561     snd @reply, $SELF;
562     };
563     };
564 root 1.3
565 root 1.8 =cut
566 root 1.3
567 root 1.22 sub psub(&) {
568     my $cb = shift;
569 root 1.3
570 root 1.22 my $port = $SELF
571     or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
572 root 1.1
573 root 1.22 sub {
574     local $SELF = $port;
575 root 1.2
576 root 1.22 if (wantarray) {
577     my @res = eval { &$cb };
578     _self_die if $@;
579     @res
580     } else {
581     my $res = eval { &$cb };
582     _self_die if $@;
583     $res
584     }
585     }
586 root 1.2 }
587    
588 root 1.67 =item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
589 root 1.32
590 root 1.67 =item $guard = mon $port, $rcvport # kill $rcvport when $port dies
591 root 1.36
592 root 1.67 =item $guard = mon $port # kill $SELF when $port dies
593 root 1.32
594 root 1.67 =item $guard = mon $port, $rcvport, @msg # send a message when $port dies
595 root 1.32
596 root 1.42 Monitor the given port and do something when the port is killed or
597     messages to it were lost, and optionally return a guard that can be used
598     to stop monitoring again.
599    
600 root 1.36 In the first form (callback), the callback is simply called with any
601     number of C<@reason> elements (no @reason means that the port was deleted
602 root 1.32 "normally"). Note also that I<< the callback B<must> never die >>, so use
603     C<eval> if unsure.
604    
605 root 1.43 In the second form (another port given), the other port (C<$rcvport>)
606 elmex 1.77 will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
607 root 1.36 "normal" kils nothing happens, while under all other conditions, the other
608     port is killed with the same reason.
609 root 1.32
610 root 1.36 The third form (kill self) is the same as the second form, except that
611     C<$rvport> defaults to C<$SELF>.
612    
613     In the last form (message), a message of the form C<@msg, @reason> will be
614     C<snd>.
615 root 1.32
616 root 1.79 Monitoring-actions are one-shot: once messages are lost (and a monitoring
617     alert was raised), they are removed and will not trigger again.
618    
619 root 1.37 As a rule of thumb, monitoring requests should always monitor a port from
620     a local port (or callback). The reason is that kill messages might get
621     lost, just like any other message. Another less obvious reason is that
622 elmex 1.77 even monitoring requests can get lost (for example, when the connection
623 root 1.37 to the other node goes down permanently). When monitoring a port locally
624     these problems do not exist.
625    
626 root 1.79 C<mon> effectively guarantees that, in the absence of hardware failures,
627     after starting the monitor, either all messages sent to the port will
628     arrive, or the monitoring action will be invoked after possible message
629     loss has been detected. No messages will be lost "in between" (after
630     the first lost message no further messages will be received by the
631     port). After the monitoring action was invoked, further messages might get
632     delivered again.
633    
634     Inter-host-connection timeouts and monitoring depend on the transport
635     used. The only transport currently implemented is TCP, and AnyEvent::MP
636     relies on TCP to detect node-downs (this can take 10-15 minutes on a
637 elmex 1.96 non-idle connection, and usually around two hours for idle connections).
638 root 1.79
639     This means that monitoring is good for program errors and cleaning up
640     stuff eventually, but they are no replacement for a timeout when you need
641     to ensure some maximum latency.
642    
643 root 1.32 Example: call a given callback when C<$port> is killed.
644    
645     mon $port, sub { warn "port died because of <@_>\n" };
646    
647     Example: kill ourselves when C<$port> is killed abnormally.
648    
649 root 1.36 mon $port;
650 root 1.32
651 root 1.36 Example: send us a restart message when another C<$port> is killed.
652 root 1.32
653     mon $port, $self => "restart";
654    
655     =cut
656    
657     sub mon {
658 root 1.75 my ($nodeid, $port) = split /#/, shift, 2;
659 root 1.32
660 root 1.75 my $node = $NODE{$nodeid} || add_node $nodeid;
661 root 1.32
662 root 1.41 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
663 root 1.32
664     unless (ref $cb) {
665     if (@_) {
666     # send a kill info message
667 root 1.41 my (@msg) = ($cb, @_);
668 root 1.32 $cb = sub { snd @msg, @_ };
669     } else {
670     # simply kill other port
671     my $port = $cb;
672     $cb = sub { kil $port, @_ if @_ };
673     }
674     }
675    
676     $node->monitor ($port, $cb);
677    
678     defined wantarray
679 root 1.124 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
680 root 1.32 }
681    
682     =item $guard = mon_guard $port, $ref, $ref...
683    
684     Monitors the given C<$port> and keeps the passed references. When the port
685     is killed, the references will be freed.
686    
687     Optionally returns a guard that will stop the monitoring.
688    
689     This function is useful when you create e.g. timers or other watchers and
690 root 1.67 want to free them when the port gets killed (note the use of C<psub>):
691 root 1.32
692     $port->rcv (start => sub {
693 root 1.67 my $timer; $timer = mon_guard $port, AE::timer 1, 1, psub {
694 root 1.32 undef $timer if 0.9 < rand;
695     });
696     });
697    
698     =cut
699    
700     sub mon_guard {
701     my ($port, @refs) = @_;
702    
703 root 1.36 #TODO: mon-less form?
704    
705 root 1.32 mon $port, sub { 0 && @refs }
706     }
707    
708 root 1.33 =item kil $port[, @reason]
709 root 1.32
710     Kill the specified port with the given C<@reason>.
711    
712 root 1.107 If no C<@reason> is specified, then the port is killed "normally" -
713     monitor callback will be invoked, but the kil will not cause linked ports
714     (C<mon $mport, $lport> form) to get killed.
715 root 1.32
716 root 1.107 If a C<@reason> is specified, then linked ports (C<mon $mport, $lport>
717     form) get killed with the same reason.
718 root 1.32
719     Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
720     will be reported as reason C<< die => $@ >>.
721    
722     Transport/communication errors are reported as C<< transport_error =>
723     $message >>.
724    
725 root 1.38 =cut
726    
727     =item $port = spawn $node, $initfunc[, @initdata]
728    
729     Creates a port on the node C<$node> (which can also be a port ID, in which
730     case it's the node where that port resides).
731    
732 root 1.67 The port ID of the newly created port is returned immediately, and it is
733     possible to immediately start sending messages or to monitor the port.
734 root 1.38
735 root 1.67 After the port has been created, the init function is called on the remote
736     node, in the same context as a C<rcv> callback. This function must be a
737     fully-qualified function name (e.g. C<MyApp::Chat::Server::init>). To
738     specify a function in the main program, use C<::name>.
739 root 1.38
740     If the function doesn't exist, then the node tries to C<require>
741     the package, then the package above the package and so on (e.g.
742     C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
743     exists or it runs out of package names.
744    
745     The init function is then called with the newly-created port as context
746 root 1.82 object (C<$SELF>) and the C<@initdata> values as arguments. It I<must>
747     call one of the C<rcv> functions to set callbacks on C<$SELF>, otherwise
748     the port might not get created.
749 root 1.38
750 root 1.67 A common idiom is to pass a local port, immediately monitor the spawned
751     port, and in the remote init function, immediately monitor the passed
752     local port. This two-way monitoring ensures that both ports get cleaned up
753     when there is a problem.
754 root 1.38
755 root 1.80 C<spawn> guarantees that the C<$initfunc> has no visible effects on the
756     caller before C<spawn> returns (by delaying invocation when spawn is
757     called for the local node).
758    
759 root 1.38 Example: spawn a chat server port on C<$othernode>.
760    
761     # this node, executed from within a port context:
762     my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
763     mon $server;
764    
765     # init function on C<$othernode>
766     sub connect {
767     my ($srcport) = @_;
768    
769     mon $srcport;
770    
771     rcv $SELF, sub {
772     ...
773     };
774     }
775    
776     =cut
777    
778     sub _spawn {
779     my $port = shift;
780     my $init = shift;
781    
782 root 1.82 # rcv will create the actual port
783 root 1.38 local $SELF = "$NODE#$port";
784     eval {
785     &{ load_func $init }
786     };
787     _self_die if $@;
788     }
789    
790     sub spawn(@) {
791 root 1.75 my ($nodeid, undef) = split /#/, shift, 2;
792 root 1.38
793 root 1.123 my $id = $RUNIQ . ++$ID;
794 root 1.38
795 root 1.39 $_[0] =~ /::/
796     or Carp::croak "spawn init function must be a fully-qualified name, caught";
797    
798 root 1.75 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
799 root 1.38
800 root 1.75 "$nodeid#$id"
801 root 1.38 }
802    
803 root 1.121
804 root 1.59 =item after $timeout, @msg
805    
806     =item after $timeout, $callback
807    
808     Either sends the given message, or call the given callback, after the
809     specified number of seconds.
810    
811 root 1.67 This is simply a utility function that comes in handy at times - the
812     AnyEvent::MP author is not convinced of the wisdom of having it, though,
813     so it may go away in the future.
814 root 1.59
815     =cut
816    
817     sub after($@) {
818     my ($timeout, @action) = @_;
819    
820     my $t; $t = AE::timer $timeout, 0, sub {
821     undef $t;
822     ref $action[0]
823     ? $action[0]()
824     : snd @action;
825     };
826     }
827    
828 root 1.87 =item cal $port, @msg, $callback[, $timeout]
829    
830     A simple form of RPC - sends a message to the given C<$port> with the
831     given contents (C<@msg>), but adds a reply port to the message.
832    
833     The reply port is created temporarily just for the purpose of receiving
834     the reply, and will be C<kil>ed when no longer needed.
835    
836     A reply message sent to the port is passed to the C<$callback> as-is.
837    
838     If an optional time-out (in seconds) is given and it is not C<undef>,
839     then the callback will be called without any arguments after the time-out
840     elapsed and the port is C<kil>ed.
841    
842 root 1.98 If no time-out is given (or it is C<undef>), then the local port will
843     monitor the remote port instead, so it eventually gets cleaned-up.
844 root 1.87
845     Currently this function returns the temporary port, but this "feature"
846     might go in future versions unless you can make a convincing case that
847     this is indeed useful for something.
848    
849     =cut
850    
851     sub cal(@) {
852     my $timeout = ref $_[-1] ? undef : pop;
853     my $cb = pop;
854    
855     my $port = port {
856     undef $timeout;
857     kil $SELF;
858     &$cb;
859     };
860    
861     if (defined $timeout) {
862     $timeout = AE::timer $timeout, 0, sub {
863     undef $timeout;
864     kil $port;
865     $cb->();
866     };
867     } else {
868     mon $_[0], sub {
869     kil $port;
870     $cb->();
871     };
872     }
873    
874     push @_, $port;
875     &snd;
876    
877     $port
878     }
879    
880 root 1.8 =back
881    
882 root 1.124 =head1 DISTRIBUTED DATABASE
883    
884     AnyEvent::MP comes with a simple distributed database. The database will
885     be mirrored asynchronously at all global nodes. Other nodes bind to one of
886     the global nodes for their needs.
887    
888     The database consists of a two-level hash - a hash contains a hash which
889     contains values.
890    
891     The top level hash key is called "family", and the second-level hash key
892     is simply called "key".
893    
894     The family and key must be alphanumeric ASCII strings, i.e. start
895     with a letter and consist of letters, digits, underscores and colons
896     (C<[A-Za-z][A-Za-z0-9_:]*>, pretty much like Perl module names.
897    
898     As the family namespaceis global, it is recommended to prefix family names
899     with the name of the application or module using it.
900    
901     The values should preferably be strings, but other perl scalars should
902     work as well (such as arrays and hashes).
903    
904     Every database entry is owned by one node - adding the same family/key
905     combination on multiple nodes will not cause discomfort for AnyEvent::MP,
906     but the result might be nondeterministic, i.e. the key might have
907     different values on different nodes.
908    
909     =item db_set $family => $key => $value
910    
911     Sets (or replaces) a key to the database.
912    
913     =item db_del $family => $key
914    
915     Deletes a key from the database.
916    
917     =item $guard = db_reg $family => $key [=> $value]
918    
919     Sets the key on the database and returns a guard. When the guard is
920     destroyed, the key is deleted from the database. If C<$value> is missing,
921     then C<undef> is used.
922    
923     =cut
924    
925     =back
926    
927 root 1.26 =head1 AnyEvent::MP vs. Distributed Erlang
928    
929 root 1.35 AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
930     == aemp node, Erlang process == aemp port), so many of the documents and
931     programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
932 root 1.27 sample:
933    
934 root 1.95 http://www.erlang.se/doc/programming_rules.shtml
935     http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
936     http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6
937     http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
938 root 1.27
939     Despite the similarities, there are also some important differences:
940 root 1.26
941     =over 4
942    
943 root 1.65 =item * Node IDs are arbitrary strings in AEMP.
944 root 1.26
945 root 1.65 Erlang relies on special naming and DNS to work everywhere in the same
946     way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
947 root 1.99 configuration or DNS), and possibly the addresses of some seed nodes, but
948     will otherwise discover other nodes (and their IDs) itself.
949 root 1.27
950 root 1.54 =item * Erlang has a "remote ports are like local ports" philosophy, AEMP
951 root 1.51 uses "local ports are like remote ports".
952    
953     The failure modes for local ports are quite different (runtime errors
954     only) then for remote ports - when a local port dies, you I<know> it dies,
955     when a connection to another node dies, you know nothing about the other
956     port.
957    
958     Erlang pretends remote ports are as reliable as local ports, even when
959     they are not.
960    
961     AEMP encourages a "treat remote ports differently" philosophy, with local
962     ports being the special case/exception, where transport errors cannot
963     occur.
964    
965 root 1.26 =item * Erlang uses processes and a mailbox, AEMP does not queue.
966    
967 root 1.119 Erlang uses processes that selectively receive messages out of order, and
968     therefore needs a queue. AEMP is event based, queuing messages would serve
969     no useful purpose. For the same reason the pattern-matching abilities
970     of AnyEvent::MP are more limited, as there is little need to be able to
971 elmex 1.77 filter messages without dequeuing them.
972 root 1.26
973 root 1.119 This is not a philosophical difference, but simply stems from AnyEvent::MP
974     being event-based, while Erlang is process-based.
975    
976     You cna have a look at L<Coro::MP> for a more Erlang-like process model on
977     top of AEMP and Coro threads.
978 root 1.26
979     =item * Erlang sends are synchronous, AEMP sends are asynchronous.
980    
981 root 1.119 Sending messages in Erlang is synchronous and blocks the process until
982     a conenction has been established and the message sent (and so does not
983     need a queue that can overflow). AEMP sends return immediately, connection
984     establishment is handled in the background.
985 root 1.26
986 root 1.51 =item * Erlang suffers from silent message loss, AEMP does not.
987 root 1.26
988 root 1.99 Erlang implements few guarantees on messages delivery - messages can get
989     lost without any of the processes realising it (i.e. you send messages a,
990     b, and c, and the other side only receives messages a and c).
991 root 1.26
992 root 1.117 AEMP guarantees (modulo hardware errors) correct ordering, and the
993     guarantee that after one message is lost, all following ones sent to the
994     same port are lost as well, until monitoring raises an error, so there are
995     no silent "holes" in the message sequence.
996 root 1.26
997 root 1.119 If you want your software to be very reliable, you have to cope with
998     corrupted and even out-of-order messages in both Erlang and AEMP. AEMP
999     simply tries to work better in common error cases, such as when a network
1000     link goes down.
1001    
1002 root 1.26 =item * Erlang can send messages to the wrong port, AEMP does not.
1003    
1004 root 1.119 In Erlang it is quite likely that a node that restarts reuses an Erlang
1005     process ID known to other nodes for a completely different process,
1006     causing messages destined for that process to end up in an unrelated
1007     process.
1008 root 1.26
1009 root 1.119 AEMP does not reuse port IDs, so old messages or old port IDs floating
1010 root 1.26 around in the network will not be sent to an unrelated port.
1011    
1012     =item * Erlang uses unprotected connections, AEMP uses secure
1013     authentication and can use TLS.
1014    
1015 root 1.66 AEMP can use a proven protocol - TLS - to protect connections and
1016 root 1.26 securely authenticate nodes.
1017    
1018 root 1.28 =item * The AEMP protocol is optimised for both text-based and binary
1019     communications.
1020    
1021 root 1.66 The AEMP protocol, unlike the Erlang protocol, supports both programming
1022 root 1.119 language independent text-only protocols (good for debugging), and binary,
1023 root 1.67 language-specific serialisers (e.g. Storable). By default, unless TLS is
1024     used, the protocol is actually completely text-based.
1025 root 1.28
1026     It has also been carefully designed to be implementable in other languages
1027 root 1.66 with a minimum of work while gracefully degrading functionality to make the
1028 root 1.28 protocol simple.
1029    
1030 root 1.35 =item * AEMP has more flexible monitoring options than Erlang.
1031    
1032 root 1.119 In Erlang, you can chose to receive I<all> exit signals as messages or
1033     I<none>, there is no in-between, so monitoring single Erlang processes is
1034     difficult to implement.
1035    
1036     Monitoring in AEMP is more flexible than in Erlang, as one can choose
1037     between automatic kill, exit message or callback on a per-port basis.
1038 root 1.35
1039 root 1.37 =item * Erlang tries to hide remote/local connections, AEMP does not.
1040 root 1.35
1041 root 1.67 Monitoring in Erlang is not an indicator of process death/crashes, in the
1042     same way as linking is (except linking is unreliable in Erlang).
1043 root 1.37
1044     In AEMP, you don't "look up" registered port names or send to named ports
1045     that might or might not be persistent. Instead, you normally spawn a port
1046 root 1.67 on the remote node. The init function monitors you, and you monitor the
1047     remote port. Since both monitors are local to the node, they are much more
1048     reliable (no need for C<spawn_link>).
1049 root 1.37
1050     This also saves round-trips and avoids sending messages to the wrong port
1051     (hard to do in Erlang).
1052 root 1.35
1053 root 1.26 =back
1054    
1055 root 1.46 =head1 RATIONALE
1056    
1057     =over 4
1058    
1059 root 1.67 =item Why strings for port and node IDs, why not objects?
1060 root 1.46
1061     We considered "objects", but found that the actual number of methods
1062 root 1.67 that can be called are quite low. Since port and node IDs travel over
1063 root 1.46 the network frequently, the serialising/deserialising would add lots of
1064 root 1.67 overhead, as well as having to keep a proxy object everywhere.
1065 root 1.46
1066     Strings can easily be printed, easily serialised etc. and need no special
1067     procedures to be "valid".
1068    
1069 root 1.110 And as a result, a port with just a default receiver consists of a single
1070 root 1.117 code reference stored in a global hash - it can't become much cheaper.
1071 root 1.47
1072 root 1.67 =item Why favour JSON, why not a real serialising format such as Storable?
1073 root 1.46
1074     In fact, any AnyEvent::MP node will happily accept Storable as framing
1075     format, but currently there is no way to make a node use Storable by
1076 root 1.67 default (although all nodes will accept it).
1077 root 1.46
1078     The default framing protocol is JSON because a) JSON::XS is many times
1079     faster for small messages and b) most importantly, after years of
1080     experience we found that object serialisation is causing more problems
1081 root 1.67 than it solves: Just like function calls, objects simply do not travel
1082 root 1.46 easily over the network, mostly because they will always be a copy, so you
1083     always have to re-think your design.
1084    
1085     Keeping your messages simple, concentrating on data structures rather than
1086     objects, will keep your messages clean, tidy and efficient.
1087    
1088     =back
1089    
1090 root 1.1 =head1 SEE ALSO
1091    
1092 root 1.68 L<AnyEvent::MP::Intro> - a gentle introduction.
1093    
1094     L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
1095    
1096 root 1.113 L<AnyEvent::MP::Global> - network maintenance and port groups, to find
1097 root 1.68 your applications.
1098    
1099 root 1.105 L<AnyEvent::MP::DataConn> - establish data connections between nodes.
1100    
1101 root 1.81 L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
1102     all nodes.
1103    
1104 root 1.1 L<AnyEvent>.
1105    
1106     =head1 AUTHOR
1107    
1108     Marc Lehmann <schmorp@schmorp.de>
1109     http://home.schmorp.de/
1110    
1111     =cut
1112    
1113     1
1114