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.47 by root, Thu Aug 13 01:57:10 2009 UTC vs.
Revision 1.140 by root, Thu Mar 22 23:47:01 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
9 $NODE # contains this node's noderef 9 $NODE # contains this node's node ID
10 NODE # returns this node's noderef 10 NODE # returns this node's node ID
11 NODE $port # returns the noderef of the port
12 11
13 $SELF # receiving/own port id in rcv callbacks 12 $SELF # receiving/own port id in rcv callbacks
14 13
14 # initialise the node so it can send/receive messages
15 configure;
16
15 # ports are message endpoints 17 # ports are message destinations
16 18
17 # sending messages 19 # sending messages
18 snd $port, type => data...; 20 snd $port, type => data...;
19 snd $port, @msg; 21 snd $port, @msg;
20 snd @msg_with_first_element_being_a_port; 22 snd @msg_with_first_element_being_a_port;
21 23
22 # miniports 24 # creating/using ports, the simple way
23 my $miniport = port { my @msg = @_; 0 }; 25 my $simple_port = port { my @msg = @_ };
24 26
25 # full ports 27 # creating/using ports, tagged message matching
26 my $port = port; 28 my $port = port;
27 rcv $port, smartmatch => $cb->(@msg);
28 rcv $port, ping => sub { snd $_[0], "pong"; 0 }; 29 rcv $port, ping => sub { snd $_[0], "pong" };
29 rcv $port, pong => sub { warn "pong received\n"; 0 }; 30 rcv $port, pong => sub { warn "pong received\n" };
30 31
31 # remote ports 32 # create a port on another node
32 my $port = spawn $node, $initfunc, @initdata; 33 my $port = spawn $node, $initfunc, @initdata;
33 34
34 # more, smarter, matches (_any_ is exported by this module) 35 # destroy a port again
35 rcv $port, [child_died => $pid] => sub { ... 36 kil $port; # "normal" kill
36 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3 37 kil $port, my_error => "everything is broken"; # error kill
37 38
38 # monitoring 39 # monitoring
39 mon $port, $cb->(@msg) # callback is invoked on death 40 mon $port, $cb->(@msg) # callback is invoked on death
40 mon $port, $otherport # kill otherport on abnormal death 41 mon $port, $localport # kill localport on abnormal death
41 mon $port, $otherport, @msg # send message on death 42 mon $port, $localport, @msg # send message on death
42 43
43=head1 CURRENT STATUS 44 # temporarily execute code in port context
45 peval $port, sub { die "kill the port!" };
44 46
45 AnyEvent::MP - stable API, should work 47 # execute callbacks in $SELF port context
46 AnyEvent::MP::Intro - outdated 48 my $timer = AE::timer 1, 0, psub {
47 AnyEvent::MP::Kernel - WIP 49 die "kill the port, delayed";
48 AnyEvent::MP::Transport - mostly stable 50 };
49
50 stay tuned.
51 51
52=head1 DESCRIPTION 52=head1 DESCRIPTION
53 53
54This module (-family) implements a simple message passing framework. 54This module (-family) implements a simple message passing framework.
55 55
56Despite its simplicity, you can securely message other processes running 56Despite its simplicity, you can securely message other processes running
57on the same or other hosts. 57on the same or other hosts, and you can supervise entities remotely.
58 58
59For an introduction to this module family, see the L<AnyEvent::MP::Intro> 59For an introduction to this module family, see the L<AnyEvent::MP::Intro>
60manual page. 60manual page and the examples under F<eg/>.
61
62At the moment, this module family is severly broken and underdocumented,
63so do not use. This was uploaded mainly to reserve the CPAN namespace -
64stay tuned!
65 61
66=head1 CONCEPTS 62=head1 CONCEPTS
67 63
68=over 4 64=over 4
69 65
70=item port 66=item port
71 67
72A port is something you can send messages to (with the C<snd> function). 68Not to be confused with a TCP port, a "port" is something you can send
69messages to (with the C<snd> function).
73 70
74Some ports allow you to register C<rcv> handlers that can match specific 71Ports allow you to register C<rcv> handlers that can match all or just
75messages. All C<rcv> handlers will receive messages they match, messages 72some messages. Messages send to ports will not be queued, regardless of
76will not be queued. 73anything was listening for them or not.
77 74
75Ports are represented by (printable) strings called "port IDs".
76
78=item port id - C<noderef#portname> 77=item port ID - C<nodeid#portname>
79 78
80A port id is normaly the concatenation of a noderef, a hash-mark (C<#>) as 79A port ID is the concatenation of a node ID, a hash-mark (C<#>)
81separator, and a port name (a printable string of unspecified format). An 80as separator, and a port name (a printable string of unspecified
82exception is the the node port, whose ID is identical to its node 81format created by AnyEvent::MP).
83reference.
84 82
85=item node 83=item node
86 84
87A node is a single process containing at least one port - the node 85A node is a single process containing at least one port - the node port,
88port. You can send messages to node ports to find existing ports or to 86which enables nodes to manage each other remotely, and to create new
89create new ports, among other things. 87ports.
90 88
91Nodes are either private (single-process only), slaves (connected to a 89Nodes are either public (have one or more listening ports) or private
92master node only) or public nodes (connectable from unrelated nodes). 90(no listening ports). Private nodes cannot talk to other private nodes
91currently, but all nodes can talk to public nodes.
93 92
94=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id> 93Nodes is represented by (printable) strings called "node IDs".
95 94
96A node reference is a string that either simply identifies the node (for 95=item node ID - C<[A-Za-z0-9_\-.:]*>
97private and slave nodes), or contains a recipe on how to reach a given
98node (for public nodes).
99 96
100This recipe is simply a comma-separated list of C<address:port> pairs (for 97A node ID is a string that uniquely identifies the node within a
101TCP/IP, other protocols might look different). 98network. Depending on the configuration used, node IDs can look like a
99hostname, a hostname and a port, or a random string. AnyEvent::MP itself
100doesn't interpret node IDs in any way except to uniquely identify a node.
102 101
103Node references come in two flavours: resolved (containing only numerical 102=item binds - C<ip:port>
104addresses) or unresolved (where hostnames are used instead of addresses).
105 103
106Before using an unresolved node reference in a message you first have to 104Nodes can only talk to each other by creating some kind of connection to
107resolve it. 105each other. To do this, nodes should listen on one or more local transport
106endpoints - binds.
107
108Currently, only standard C<ip:port> specifications can be used, which
109specify TCP ports to listen on. So a bind is basically just a tcp socket
110in listening mode thta accepts conenctions form other nodes.
111
112=item seed nodes
113
114When a node starts, it knows nothing about the network it is in - it
115needs to connect to at least one other node that is already in the
116network. These other nodes are called "seed nodes".
117
118Seed nodes themselves are not special - they are seed nodes only because
119some other node I<uses> them as such, but any node can be used as seed
120node for other nodes, and eahc node cna use a different set of seed nodes.
121
122In addition to discovering the network, seed nodes are also used to
123maintain the network - all nodes using the same seed node form are part of
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.
127
128Seed nodes are expected to be long-running, and at least one seed node
129should always be available. They should also be relatively responsive - a
130seed node that blocks for long periods will slow down everybody else.
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
141=item seed IDs - C<host:port>
142
143Seed IDs are transport endpoint(s) (usually a hostname/IP address and a
144TCP port) of nodes that should be used as seed nodes.
145
146=item global nodes
147
148An AEMP network needs a discovery service - nodes need to know how to
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).
108 162
109=back 163=back
110 164
111=head1 VARIABLES/FUNCTIONS 165=head1 VARIABLES/FUNCTIONS
112 166
114 168
115=cut 169=cut
116 170
117package AnyEvent::MP; 171package AnyEvent::MP;
118 172
173use AnyEvent::MP::Config ();
119use AnyEvent::MP::Kernel; 174use AnyEvent::MP::Kernel;
175use AnyEvent::MP::Kernel qw(%NODE %PORT %PORT_DATA $UNIQ $RUNIQ $ID);
120 176
121use common::sense; 177use common::sense;
122 178
123use Carp (); 179use Carp ();
124 180
125use AE (); 181use AE ();
182use Guard ();
126 183
127use base "Exporter"; 184use base "Exporter";
128 185
129our $VERSION = $AnyEvent::MP::Kernel::VERSION; 186our $VERSION = $AnyEvent::MP::Config::VERSION;
130 187
131our @EXPORT = qw( 188our @EXPORT = qw(
132 NODE $NODE *SELF node_of _any_ 189 NODE $NODE *SELF node_of after
133 resolve_node initialise_node 190 configure
134 snd rcv mon kil reg psub spawn 191 snd rcv mon mon_guard kil psub peval spawn cal
135 port 192 port
193 db_set db_del db_reg
194 db_mon db_family db_keys db_values
136); 195);
137 196
138our $SELF; 197our $SELF;
139 198
140sub _self_die() { 199sub _self_die() {
143 kil $SELF, die => $msg; 202 kil $SELF, die => $msg;
144} 203}
145 204
146=item $thisnode = NODE / $NODE 205=item $thisnode = NODE / $NODE
147 206
148The C<NODE> function returns, and the C<$NODE> variable contains 207The C<NODE> function returns, and the C<$NODE> variable contains, the node
149the noderef of the local node. The value is initialised by a call 208ID of the node running in the current process. This value is initialised by
150to C<become_public> or C<become_slave>, after which all local port 209a call to C<configure>.
151identifiers become invalid.
152 210
153=item $noderef = node_of $port 211=item $nodeid = node_of $port
154 212
155Extracts and returns the noderef from a portid or a noderef. 213Extracts and returns the node ID from a port ID or a node ID.
156 214
157=item initialise_node $noderef, $seednode, $seednode... 215=item configure $profile, key => value...
158 216
159=item initialise_node "slave/", $master, $master... 217=item configure key => value...
160 218
161Before a node can talk to other nodes on the network it has to initialise 219Before a node can talk to other nodes on the network (i.e. enter
162itself - the minimum a node needs to know is it's own name, and optionally 220"distributed mode") it has to configure itself - the minimum a node needs
163it should know the noderefs of some other nodes in the network. 221to know is its own name, and optionally it should know the addresses of
222some other nodes in the network to discover other nodes.
164 223
165This function initialises a node - it must be called exactly once (or 224This function configures a node - it must be called exactly once (or
166never) before calling other AnyEvent::MP functions. 225never) before calling other AnyEvent::MP functions.
167 226
168All arguments are noderefs, which can be either resolved or unresolved. 227The key/value pairs are basically the same ones as documented for the
169 228F<aemp> command line utility (sans the set/del prefix), with these additions:
170There are two types of networked nodes, public nodes and slave nodes:
171 229
172=over 4 230=over 4
173 231
174=item public nodes 232=item norc => $boolean (default false)
175 233
176For public nodes, C<$noderef> must either be a (possibly unresolved) 234If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not>
177noderef, in which case it will be resolved, or C<undef> (or missing), in 235be consulted - all configuraiton options must be specified in the
178which case the noderef will be guessed. 236C<configure> call.
179 237
180Afterwards, the node will bind itself on all endpoints and try to connect 238=item force => $boolean (default false)
181to all additional C<$seednodes> that are specified. Seednodes are optional
182and can be used to quickly bootstrap the node into an existing network.
183 239
184=item slave nodes 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.
185 243
186When the C<$noderef> is the special string C<slave/>, then the node will 244=item secure => $pass->(@msg)
187become a slave node. Slave nodes cannot be contacted from outside and will
188route most of their traffic to the master node that they attach to.
189 245
190At least one additional noderef is required: The node will try to connect 246In addition to specifying a boolean, you can specify a code reference that
191to all of them and will become a slave attached to the first node it can 247is called for every code execution attempt - the execution request is
192successfully connect to. 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.
193 256
194=back 257=back
195 258
196This function will block until all nodes have been resolved and, for slave
197nodes, until it has successfully established a connection to a master
198server.
199
200Example: become a public node listening on the default node.
201
202 initialise_node;
203
204Example: become a public node, and try to contact some well-known master
205servers to become part of the network.
206
207 initialise_node undef, "master1", "master2";
208
209Example: become a public node listening on port C<4041>.
210
211 initialise_node 4041;
212
213Example: become a public node, only visible on localhost port 4044.
214
215 initialise_node "locahost:4044";
216
217Example: become a slave node to any of the specified master servers.
218
219 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
220
221=item $cv = resolve_node $noderef
222
223Takes an unresolved node reference that may contain hostnames and
224abbreviated IDs, resolves all of them and returns a resolved node
225reference.
226
227In addition to C<address:port> pairs allowed in resolved noderefs, the
228following forms are supported:
229
230=over 4 259=over 4
231 260
232=item the empty string 261=item step 1, gathering configuration from profiles
233 262
234An empty-string component gets resolved as if the default port (4040) was 263The function first looks up a profile in the aemp configuration (see the
235specified. 264L<aemp> commandline utility). The profile name can be specified via the
265named C<profile> parameter or can simply be the first parameter). If it is
266missing, then the nodename (F<uname -n>) will be used as profile name.
236 267
237=item naked port numbers (e.g. C<1234>) 268The profile data is then gathered as follows:
238 269
239These are resolved by prepending the local nodename and a colon, to be 270First, all remaining key => value pairs (all of which are conveniently
240further resolved. 271undocumented at the moment) will be interpreted as configuration
272data. Then they will be overwritten by any values specified in the global
273default configuration (see the F<aemp> utility), then the chain of
274profiles chosen by the profile name (and any C<parent> attributes).
241 275
242=item hostnames (e.g. C<localhost:1234>, C<localhost>) 276That means that the values specified in the profile have highest priority
277and the values specified directly via C<configure> have lowest priority,
278and can only be used to specify defaults.
243 279
244These are resolved by using AnyEvent::DNS to resolve them, optionally 280If the profile specifies a node ID, then this will become the node ID of
245looking up SRV records for the C<aemp=4040> port, if no port was 281this process. If not, then the profile name will be used as node ID, with
246specified. 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>.
289
290=item step 2, bind listener sockets
291
292The next step is to look up the binds in the profile, followed by binding
293aemp protocol listeners on all binds specified (it is possible and valid
294to have no binds, meaning that the node cannot be contacted form the
295outside. This means the node cannot talk to other nodes that also have no
296binds, but it can still talk to all "normal" nodes).
297
298If the profile does not specify a binds list, then a default of C<*> is
299used, meaning the node will bind on a dynamically-assigned port on every
300local IP address it finds.
301
302=item step 3, connect to seed nodes
303
304As the last step, the seed ID list from the profile is passed to the
305L<AnyEvent::MP::Global> module, which will then use it to keep
306connectivity with at least one node at any point in time.
247 307
248=back 308=back
309
310Example: become a distributed node using the local node name as profile.
311This should be the most common form of invocation for "daemon"-type nodes.
312
313 configure
314
315Example: become a semi-anonymous node. This form is often used for
316commandline clients.
317
318 configure nodeid => "myscript/%n/%u";
319
320Example: configure a node using a profile called seed, which is suitable
321for a seed node as it binds on all local addresses on a fixed port (4040,
322customary for aemp).
323
324 # use the aemp commandline utility
325 # aemp profile seed binds '*:4040'
326
327 # then use it
328 configure profile => "seed";
329
330 # or simply use aemp from the shell again:
331 # aemp run profile seed
332
333 # or provide a nicer-to-remember nodeid
334 # aemp run profile seed nodeid "$(hostname)"
249 335
250=item $SELF 336=item $SELF
251 337
252Contains the current port id while executing C<rcv> callbacks or C<psub> 338Contains the current port id while executing C<rcv> callbacks or C<psub>
253blocks. 339blocks.
254 340
255=item SELF, %SELF, @SELF... 341=item *SELF, SELF, %SELF, @SELF...
256 342
257Due to some quirks in how perl exports variables, it is impossible to 343Due to some quirks in how perl exports variables, it is impossible to
258just export C<$SELF>, all the symbols called C<SELF> are exported by this 344just export C<$SELF>, all the symbols named C<SELF> are exported by this
259module, but only C<$SELF> is currently used. 345module, but only C<$SELF> is currently used.
260 346
261=item snd $port, type => @data 347=item snd $port, type => @data
262 348
263=item snd $port, @msg 349=item snd $port, @msg
264 350
265Send the given message to the given port ID, which can identify either 351Send the given message to the given port, which can identify either a
266a local or a remote port, and can be either a string or soemthignt hat 352local or a remote port, and must be a port ID.
267stringifies a sa port ID (such as a port object :).
268 353
269While the message can be about anything, it is highly recommended to use a 354While the message can be almost anything, it is highly recommended to
270string as first element (a portid, or some word that indicates a request 355use a string as first element (a port ID, or some word that indicates a
271type etc.). 356request type etc.) and to consist if only simple perl values (scalars,
357arrays, hashes) - if you think you need to pass an object, think again.
272 358
273The message data effectively becomes read-only after a call to this 359The message data logically becomes read-only after a call to this
274function: modifying any argument is not allowed and can cause many 360function: modifying any argument (or values referenced by them) is
275problems. 361forbidden, as there can be considerable time between the call to C<snd>
362and the time the message is actually being serialised - in fact, it might
363never be copied as within the same process it is simply handed to the
364receiving port.
276 365
277The type of data you can transfer depends on the transport protocol: when 366The type of data you can transfer depends on the transport protocol: when
278JSON is used, then only strings, numbers and arrays and hashes consisting 367JSON is used, then only strings, numbers and arrays and hashes consisting
279of those are allowed (no objects). When Storable is used, then anything 368of those are allowed (no objects). When Storable is used, then anything
280that Storable can serialise and deserialise is allowed, and for the local 369that Storable can serialise and deserialise is allowed, and for the local
281node, anything can be passed. 370node, anything can be passed. Best rely only on the common denominator of
371these.
282 372
283=item $local_port = port 373=item $local_port = port
284 374
285Create a new local port object that can be used either as a pattern 375Create a new local port object and returns its port ID. Initially it has
286matching port ("full port") or a single-callback port ("miniport"), 376no callbacks set and will throw an error when it receives messages.
287depending on how C<rcv> callbacks are bound to the object.
288 377
289=item $port = port { my @msg = @_; $finished } 378=item $local_port = port { my @msg = @_ }
290 379
291Creates a "miniport", that is, a very lightweight port without any pattern 380Creates a new local port, and returns its ID. Semantically the same as
292matching behind it, and returns its ID. Semantically the same as creating
293a port and calling C<rcv $port, $callback> on it. 381creating a port and calling C<rcv $port, $callback> on it.
294 382
295The block will be called for every message received on the port. When the 383The block will be called for every message received on the port, with the
296callback returns a true value its job is considered "done" and the port 384global variable C<$SELF> set to the port ID. Runtime errors will cause the
297will be destroyed. Otherwise it will stay alive. 385port to be C<kil>ed. The message will be passed as-is, no extra argument
386(i.e. no port ID) will be passed to the callback.
298 387
299The message will be passed as-is, no extra argument (i.e. no port id) will 388If you want to stop/destroy the port, simply C<kil> it:
300be passed to the callback.
301 389
302If you need the local port id in the callback, this works nicely: 390 my $port = port {
303 391 my @msg = @_;
304 my $port; $port = port { 392 ...
305 snd $otherport, reply => $port; 393 kil $SELF;
306 }; 394 };
307 395
308=cut 396=cut
309 397
310sub rcv($@); 398sub rcv($@);
311 399
400my $KILME = sub {
401 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
402 kil $SELF, unhandled_message => "no callback found for message '$tag'";
403};
404
312sub port(;&) { 405sub port(;&) {
313 my $id = "$UNIQ." . $ID++; 406 my $id = $UNIQ . ++$ID;
314 my $port = "$NODE#$id"; 407 my $port = "$NODE#$id";
315 408
316 if (@_) {
317 rcv $port, shift; 409 rcv $port, shift || $KILME;
318 } else {
319 $PORT{$id} = sub { }; # nop
320 }
321 410
322 $port 411 $port
323} 412}
324 413
325=item reg $port, $name
326
327=item reg $name
328
329Registers the given port (or C<$SELF><<< if missing) under the name
330C<$name>. If the name already exists it is replaced.
331
332A port can only be registered under one well known name.
333
334A port automatically becomes unregistered when it is killed.
335
336=cut
337
338sub reg(@) {
339 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
340
341 $REG{$_[0]} = $port;
342}
343
344=item rcv $port, $callback->(@msg) 414=item rcv $local_port, $callback->(@msg)
345 415
346Replaces the callback on the specified miniport (after converting it to 416Replaces the default callback on the specified port. There is no way to
347one if required). 417remove the default callback: use C<sub { }> to disable it, or better
348 418C<kil> the port when it is no longer needed.
349=item rcv $port, tagstring => $callback->(@msg), ...
350
351=item rcv $port, $smartmatch => $callback->(@msg), ...
352
353=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
354
355Register callbacks to be called on matching messages on the given full
356port (after converting it to one if required) and return the port.
357
358The callback has to return a true value when its work is done, after
359which is will be removed, or a false value in which case it will stay
360registered.
361 419
362The global C<$SELF> (exported by this module) contains C<$port> while 420The global C<$SELF> (exported by this module) contains C<$port> while
363executing the callback. 421executing the callback. Runtime errors during callback execution will
422result in the port being C<kil>ed.
364 423
365Runtime errors during callback execution will result in the port being 424The default callback receives all messages not matched by a more specific
366C<kil>ed. 425C<tag> match.
367 426
368If the match is an array reference, then it will be matched against the 427=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
369first elements of the message, otherwise only the first element is being
370matched.
371 428
372Any element in the match that is specified as C<_any_> (a function 429Register (or replace) callbacks to be called on messages starting with the
373exported by this module) matches any single element of the message. 430given tag on the given port (and return the port), or unregister it (when
431C<$callback> is C<$undef> or missing). There can only be one callback
432registered for each tag.
374 433
375While not required, it is highly recommended that the first matching 434The original message will be passed to the callback, after the first
376element is a string identifying the message. The one-string-only match is 435element (the tag) has been removed. The callback will use the same
377also the most efficient match (by far). 436environment as the default callback (see above).
378 437
379Example: create a port and bind receivers on it in one go. 438Example: create a port and bind receivers on it in one go.
380 439
381 my $port = rcv port, 440 my $port = rcv port,
382 msg1 => sub { ...; 0 }, 441 msg1 => sub { ... },
383 msg2 => sub { ...; 0 }, 442 msg2 => sub { ... },
384 ; 443 ;
385 444
386Example: create a port, bind receivers and send it in a message elsewhere 445Example: create a port, bind receivers and send it in a message elsewhere
387in one go: 446in one go:
388 447
389 snd $otherport, reply => 448 snd $otherport, reply =>
390 rcv port, 449 rcv port,
391 msg1 => sub { ...; 0 }, 450 msg1 => sub { ... },
392 ... 451 ...
393 ; 452 ;
394 453
454Example: temporarily register a rcv callback for a tag matching some port
455(e.g. for an rpc reply) and unregister it after a message was received.
456
457 rcv $port, $otherport => sub {
458 my @reply = @_;
459
460 rcv $SELF, $otherport;
461 };
462
395=cut 463=cut
396 464
397sub rcv($@) { 465sub rcv($@) {
398 my $port = shift; 466 my $port = shift;
399 my ($noderef, $portid) = split /#/, $port, 2; 467 my ($nodeid, $portid) = split /#/, $port, 2;
400 468
401 ($NODE{$noderef} || add_node $noderef) == $NODE{""} 469 $NODE{$nodeid} == $NODE{""}
402 or Carp::croak "$port: rcv can only be called on local ports, caught"; 470 or Carp::croak "$port: rcv can only be called on local ports, caught";
403 471
404 if (@_ == 1) { 472 while (@_) {
473 if (ref $_[0]) {
474 if (my $self = $PORT_DATA{$portid}) {
475 "AnyEvent::MP::Port" eq ref $self
476 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
477
478 $self->[0] = shift;
479 } else {
405 my $cb = shift; 480 my $cb = shift;
406 delete $PORT_DATA{$portid};
407 $PORT{$portid} = sub { 481 $PORT{$portid} = sub {
408 local $SELF = $port; 482 local $SELF = $port;
409 eval { 483 eval { &$cb }; _self_die if $@;
410 &$cb 484 };
411 and kil $port;
412 }; 485 }
413 _self_die if $@; 486 } elsif (defined $_[0]) {
414 };
415 } else {
416 my $self = $PORT_DATA{$portid} ||= do { 487 my $self = $PORT_DATA{$portid} ||= do {
417 my $self = bless { 488 my $self = bless [$PORT{$portid} || sub { }, { }, $port], "AnyEvent::MP::Port";
418 id => $port,
419 }, "AnyEvent::MP::Port";
420 489
421 $PORT{$portid} = sub { 490 $PORT{$portid} = sub {
422 local $SELF = $port; 491 local $SELF = $port;
423 492
424 eval {
425 for (@{ $self->{rc0}{$_[0]} }) { 493 if (my $cb = $self->[1]{$_[0]}) {
426 $_ && &{$_->[0]} 494 shift;
427 && undef $_; 495 eval { &$cb }; _self_die if $@;
428 } 496 } else {
429
430 for (@{ $self->{rcv}{$_[0]} }) {
431 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
432 && &{$_->[0]} 497 &{ $self->[0] };
433 && undef $_;
434 }
435
436 for (@{ $self->{any} }) {
437 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
438 && &{$_->[0]}
439 && undef $_;
440 } 498 }
441 }; 499 };
442 _self_die if $@; 500
501 $self
443 }; 502 };
444 503
445 $self
446 };
447
448 "AnyEvent::MP::Port" eq ref $self 504 "AnyEvent::MP::Port" eq ref $self
449 or Carp::croak "$port: rcv can only be called on message matching ports, caught"; 505 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
450 506
451 while (@_) {
452 my ($match, $cb) = splice @_, 0, 2; 507 my ($tag, $cb) = splice @_, 0, 2;
453 508
454 if (!ref $match) { 509 if (defined $cb) {
455 push @{ $self->{rc0}{$match} }, [$cb]; 510 $self->[1]{$tag} = $cb;
456 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
457 my ($type, @match) = @$match;
458 @match
459 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
460 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
461 } else { 511 } else {
462 push @{ $self->{any} }, [$cb, $match]; 512 delete $self->[1]{$tag};
463 } 513 }
464 } 514 }
465 } 515 }
466 516
467 $port 517 $port
468} 518}
469 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
470=item $closure = psub { BLOCK } 557=item $closure = psub { BLOCK }
471 558
472Remembers 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
473closure 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>
474callbacks, 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 }, @_ } >>.
475 565
476This is useful when you register callbacks from C<rcv> callbacks: 566This is useful when you register callbacks from C<rcv> callbacks:
477 567
478 rcv delayed_reply => sub { 568 rcv delayed_reply => sub {
479 my ($delay, @reply) = @_; 569 my ($delay, @reply) = @_;
503 $res 593 $res
504 } 594 }
505 } 595 }
506} 596}
507 597
508=item $guard = mon $port, $cb->(@reason) 598=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
509 599
510=item $guard = mon $port, $rcvport 600=item $guard = mon $port # kill $SELF when $port dies
511 601
512=item $guard = mon $port 602=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
513 603
514=item $guard = mon $port, $rcvport, @msg 604=item $guard = mon $port, $rcvport, @msg # send a message when $port dies
515 605
516Monitor 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
517messages 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
518to stop monitoring again. 608to stop monitoring again.
519 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
624In the third form (callback), the callback is simply called with any
625number of C<@reason> elements (empty @reason means that the port was deleted
626"normally"). Note also that I<< the callback B<must> never die >>, so use
627C<eval> if unsure.
628
629In the last form (message), a message of the form C<$rcvport, @msg,
630@reason> will be C<snd>.
631
632Monitoring-actions are one-shot: once messages are lost (and a monitoring
633alert was raised), they are removed and will not trigger again, even if it
634turns out that the port is still alive.
635
636As a rule of thumb, monitoring requests should always monitor a remote
637port locally (using a local C<$rcvport> or a callback). The reason is that
638kill messages might get lost, just like any other message. Another less
639obvious reason is that even monitoring requests can get lost (for example,
640when the connection to the other node goes down permanently). When
641monitoring a port locally these problems do not exist.
642
520C<mon> effectively guarantees that, in the absence of hardware failures, 643C<mon> effectively guarantees that, in the absence of hardware failures,
521that after starting the monitor, either all messages sent to the port 644after starting the monitor, either all messages sent to the port will
522will arrive, or the monitoring action will be invoked after possible 645arrive, or the monitoring action will be invoked after possible message
523message loss has been detected. No messages will be lost "in between" 646loss has been detected. No messages will be lost "in between" (after
524(after the first lost message no further messages will be received by the 647the first lost message no further messages will be received by the
525port). After the monitoring action was invoked, further messages might get 648port). After the monitoring action was invoked, further messages might get
526delivered again. 649delivered again.
527 650
528In the first form (callback), the callback is simply called with any 651Inter-host-connection timeouts and monitoring depend on the transport
529number of C<@reason> elements (no @reason means that the port was deleted 652used. The only transport currently implemented is TCP, and AnyEvent::MP
530"normally"). Note also that I<< the callback B<must> never die >>, so use 653relies on TCP to detect node-downs (this can take 10-15 minutes on a
531C<eval> if unsure. 654non-idle connection, and usually around two hours for idle connections).
532 655
533In the second form (another port given), the other port (C<$rcvport>) 656This means that monitoring is good for program errors and cleaning up
534will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on 657stuff eventually, but they are no replacement for a timeout when you need
535"normal" kils nothing happens, while under all other conditions, the other 658to ensure some maximum latency.
536port is killed with the same reason.
537
538The third form (kill self) is the same as the second form, except that
539C<$rvport> defaults to C<$SELF>.
540
541In the last form (message), a message of the form C<@msg, @reason> will be
542C<snd>.
543
544As a rule of thumb, monitoring requests should always monitor a port from
545a local port (or callback). The reason is that kill messages might get
546lost, just like any other message. Another less obvious reason is that
547even monitoring requests can get lost (for exmaple, when the connection
548to the other node goes down permanently). When monitoring a port locally
549these problems do not exist.
550 659
551Example: call a given callback when C<$port> is killed. 660Example: call a given callback when C<$port> is killed.
552 661
553 mon $port, sub { warn "port died because of <@_>\n" }; 662 mon $port, sub { warn "port died because of <@_>\n" };
554 663
561 mon $port, $self => "restart"; 670 mon $port, $self => "restart";
562 671
563=cut 672=cut
564 673
565sub mon { 674sub mon {
566 my ($noderef, $port) = split /#/, shift, 2; 675 my ($nodeid, $port) = split /#/, shift, 2;
567 676
568 my $node = $NODE{$noderef} || add_node $noderef; 677 my $node = $NODE{$nodeid} || add_node $nodeid;
569 678
570 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,'; 679 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
571 680
572 unless (ref $cb) { 681 unless (ref $cb) {
573 if (@_) { 682 if (@_) {
582 } 691 }
583 692
584 $node->monitor ($port, $cb); 693 $node->monitor ($port, $cb);
585 694
586 defined wantarray 695 defined wantarray
587 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) } 696 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
588} 697}
589 698
590=item $guard = mon_guard $port, $ref, $ref... 699=item $guard = mon_guard $port, $ref, $ref...
591 700
592Monitors 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
593is killed, the references will be freed. 702is killed, the references will be freed.
594 703
595Optionally returns a guard that will stop the monitoring. 704Optionally returns a guard that will stop the monitoring.
596 705
597This function is useful when you create e.g. timers or other watchers and 706This function is useful when you create e.g. timers or other watchers and
598want to free them when the port gets killed: 707want to free them when the port gets killed (note the use of C<psub>):
599 708
600 $port->rcv (start => sub { 709 $port->rcv (start => sub {
601 my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub { 710 my $timer; $timer = mon_guard $port, AE::timer 1, 1, psub {
602 undef $timer if 0.9 < rand; 711 undef $timer if 0.9 < rand;
603 }); 712 });
604 }); 713 });
605 714
606=cut 715=cut
615 724
616=item kil $port[, @reason] 725=item kil $port[, @reason]
617 726
618Kill the specified port with the given C<@reason>. 727Kill the specified port with the given C<@reason>.
619 728
620If no C<@reason> is specified, then the port is killed "normally" (linked 729If no C<@reason> is specified, then the port is killed "normally" -
621ports will not be kileld, or even notified). 730monitor callback will be invoked, but the kil will not cause linked ports
731(C<mon $mport, $lport> form) to get killed.
622 732
623Otherwise, linked ports get killed with the same reason (second form of 733If a C<@reason> is specified, then linked ports (C<mon $mport, $lport>
624C<mon>, see below). 734form) get killed with the same reason.
625 735
626Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 736Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
627will be reported as reason C<< die => $@ >>. 737will be reported as reason C<< die => $@ >>.
628 738
629Transport/communication errors are reported as C<< transport_error => 739Transport/communication errors are reported as C<< transport_error =>
630$message >>. 740$message >>.
631 741
632=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: $!";
633 753
634=item $port = spawn $node, $initfunc[, @initdata] 754=item $port = spawn $node, $initfunc[, @initdata]
635 755
636Creates 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
637case it's the node where that port resides). 757case it's the node where that port resides).
638 758
639The port ID of the newly created port is return immediately, and it is 759The port ID of the newly created port is returned immediately, and it is
640permissible to immediately start sending messages or monitor the port. 760possible to immediately start sending messages or to monitor the port.
641 761
642After the port has been created, the init function is 762After the port has been created, the init function is called on the remote
643called. This function must be a fully-qualified function name 763node, in the same context as a C<rcv> callback. This function must be a
644(e.g. C<MyApp::Chat::Server::init>). To specify a function in the main 764fully-qualified function name (e.g. C<MyApp::Chat::Server::init>). To
645program, use C<::name>. 765specify a function in the main program, use C<::name>.
646 766
647If the function doesn't exist, then the node tries to C<require> 767If the function doesn't exist, then the node tries to C<require>
648the package, then the package above the package and so on (e.g. 768the package, then the package above the package and so on (e.g.
649C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function 769C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
650exists or it runs out of package names. 770exists or it runs out of package names.
651 771
652The init function is then called with the newly-created port as context 772The init function is then called with the newly-created port as context
653object (C<$SELF>) and the C<@initdata> values as arguments. 773object (C<$SELF>) and the C<@initdata> values as arguments. It I<must>
774call one of the C<rcv> functions to set callbacks on C<$SELF>, otherwise
775the port might not get created.
654 776
655A common idiom is to pass your own port, monitor the spawned port, and 777A common idiom is to pass a local port, immediately monitor the spawned
656in the init function, monitor the original port. This two-way monitoring 778port, and in the remote init function, immediately monitor the passed
657ensures that both ports get cleaned up when there is a problem. 779local port. This two-way monitoring ensures that both ports get cleaned up
780when there is a problem.
781
782C<spawn> guarantees that the C<$initfunc> has no visible effects on the
783caller before C<spawn> returns (by delaying invocation when spawn is
784called for the local node).
658 785
659Example: spawn a chat server port on C<$othernode>. 786Example: spawn a chat server port on C<$othernode>.
660 787
661 # this node, executed from within a port context: 788 # this node, executed from within a port context:
662 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF; 789 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
677 804
678sub _spawn { 805sub _spawn {
679 my $port = shift; 806 my $port = shift;
680 my $init = shift; 807 my $init = shift;
681 808
809 # rcv will create the actual port
682 local $SELF = "$NODE#$port"; 810 local $SELF = "$NODE#$port";
683 eval { 811 eval {
684 &{ load_func $init } 812 &{ load_func $init }
685 }; 813 };
686 _self_die if $@; 814 _self_die if $@;
687} 815}
688 816
689sub spawn(@) { 817sub spawn(@) {
690 my ($noderef, undef) = split /#/, shift, 2; 818 my ($nodeid, undef) = split /#/, shift, 2;
691 819
692 my $id = "$RUNIQ." . $ID++; 820 my $id = $RUNIQ . ++$ID;
693 821
694 $_[0] =~ /::/ 822 $_[0] =~ /::/
695 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";
696 824
697 ($NODE{$noderef} || add_node $noderef) 825 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
698 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
699 826
700 "$noderef#$id" 827 "$nodeid#$id"
701} 828}
702 829
830
831=item after $timeout, @msg
832
833=item after $timeout, $callback
834
835Either sends the given message, or call the given callback, after the
836specified number of seconds.
837
838This is simply a utility function that comes in handy at times - the
839AnyEvent::MP author is not convinced of the wisdom of having it, though,
840so it may go away in the future.
841
842=cut
843
844sub after($@) {
845 my ($timeout, @action) = @_;
846
847 my $t; $t = AE::timer $timeout, 0, sub {
848 undef $t;
849 ref $action[0]
850 ? $action[0]()
851 : snd @action;
852 };
853}
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
703=back 909=back
704 910
705=head1 NODE MESSAGES 911=head1 DISTRIBUTED DATABASE
706 912
707Nodes understand the following messages sent to them. Many of them take 913AnyEvent::MP comes with a simple distributed database. The database will
708arguments called C<@reply>, which will simply be used to compose a reply 914be mirrored asynchronously on all global nodes. Other nodes bind to one
709message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and 915of the global nodes for their needs. Every node has a "local database"
710the remaining arguments are simply the message data. 916which contains all the values that are set locally. All local databases
917are merged together to form the global database, which can be queried.
711 918
712While other messages exist, they are not public and subject to change. 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.:
713 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
714=over 4 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.
715 1007
716=cut 1008=cut
717 1009
718=item lookup => $name, @reply 1010sub db_reg($$;$) {
1011 my $family = shift;
1012 my $port = @_ ? shift : $SELF;
719 1013
720Replies with the port ID of the specified well-known port, or C<undef>. 1014 my $clr = sub { db_del $family => $port };
1015 mon $port, $clr;
721 1016
722=item devnull => ... 1017 db_set $family => $port => $_[0];
723 1018
724Generic data sink/CPU heat conversion. 1019 defined wantarray
1020 and &Guard::guard ($clr)
1021}
725 1022
726=item relay => $port, @msg 1023=item db_family $family => $cb->(\%familyhash)
727 1024
728Simply forwards the message to the given port. 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.
729 1027
730=item eval => $string[ @reply] 1028=item db_keys $family => $cb->(\@keys)
731 1029
732Evaluates the given string. If C<@reply> is given, then a message of the 1030Same as C<db_family>, except it only queries the family I<subkeys> and passes
733form C<@reply, $@, @evalres> is sent. 1031them as array reference to the callback.
734 1032
735Example: crash another node. 1033=item db_values $family => $cb->(\@values)
736 1034
737 snd $othernode, eval => "exit"; 1035Same as C<db_family>, except it only queries the family I<values> and passes them
1036as array reference to the callback.
738 1037
739=item time => @reply 1038=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
740 1039
741Replies the the current node time to C<@reply>. 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.
742 1045
743Example: tell the current node to send the current time to C<$myport> in a 1046If not called in void context, a guard object is returned that, when
744C<timereply> message. 1047destroyed, stops the monitor.
745 1048
746 snd $NODE, time => $myport, timereply => 1, 2; 1049The family hash reference and the key arrays belong to AnyEvent::MP and
747 # => snd $myport, timereply => 1, 2, <time> 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
748 1090
749=back 1091=back
750 1092
751=head1 AnyEvent::MP vs. Distributed Erlang 1093=head1 AnyEvent::MP vs. Distributed Erlang
752 1094
753AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 1095AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
754== 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
755programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 1097programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
756sample: 1098sample:
757 1099
758 http://www.Erlang.se/doc/programming_rules.shtml 1100 http://www.erlang.se/doc/programming_rules.shtml
759 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
760 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
761 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
762 1104
763Despite the similarities, there are also some important differences: 1105Despite the similarities, there are also some important differences:
764 1106
765=over 4 1107=over 4
766 1108
767=item * Node references contain the recipe on how to contact them. 1109=item * Node IDs are arbitrary strings in AEMP.
768 1110
769Erlang relies on special naming and DNS to work everywhere in the 1111Erlang relies on special naming and DNS to work everywhere in the same
770same way. AEMP relies on each node knowing it's own address(es), with 1112way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
771convenience functionality. 1113configuration or DNS), and possibly the addresses of some seed nodes, but
1114will otherwise discover other nodes (and their IDs) itself.
772 1115
773This means that AEMP requires a less tightly controlled environment at the 1116=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
774cost of longer node references and a slightly higher management overhead. 1117uses "local ports are like remote ports".
1118
1119The failure modes for local ports are quite different (runtime errors
1120only) then for remote ports - when a local port dies, you I<know> it dies,
1121when a connection to another node dies, you know nothing about the other
1122port.
1123
1124Erlang pretends remote ports are as reliable as local ports, even when
1125they are not.
1126
1127AEMP encourages a "treat remote ports differently" philosophy, with local
1128ports being the special case/exception, where transport errors cannot
1129occur.
775 1130
776=item * Erlang uses processes and a mailbox, AEMP does not queue. 1131=item * Erlang uses processes and a mailbox, AEMP does not queue.
777 1132
778Erlang uses processes that selctively receive messages, and therefore 1133Erlang uses processes that selectively receive messages out of order, and
779needs a queue. AEMP is event based, queuing messages would serve no useful 1134therefore needs a queue. AEMP is event based, queuing messages would serve
780purpose. 1135no useful purpose. For the same reason the pattern-matching abilities
1136of AnyEvent::MP are more limited, as there is little need to be able to
1137filter messages without dequeuing them.
781 1138
782(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.
783 1144
784=item * Erlang sends are synchronous, AEMP sends are asynchronous. 1145=item * Erlang sends are synchronous, AEMP sends are asynchronous.
785 1146
786Sending messages in Erlang is synchronous and blocks the process. AEMP 1147Sending messages in Erlang is synchronous and blocks the process until
787sends are immediate, connection establishment is handled in the 1148a conenction has been established and the message sent (and so does not
788background. 1149need a queue that can overflow). AEMP sends return immediately, connection
1150establishment is handled in the background.
789 1151
790=item * Erlang can silently lose messages, AEMP cannot. 1152=item * Erlang suffers from silent message loss, AEMP does not.
791 1153
792Erlang makes few guarantees on messages delivery - messages can get lost 1154Erlang implements few guarantees on messages delivery - messages can get
793without 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,
794and c, and the other side only receives messages a and c). 1156b, and c, and the other side only receives messages a and c).
795 1157
796AEMP guarantees correct ordering, and the guarantee that there are no 1158AEMP guarantees (modulo hardware errors) correct ordering, and the
1159guarantee that after one message is lost, all following ones sent to the
1160same port are lost as well, until monitoring raises an error, so there are
797holes in the message sequence. 1161no silent "holes" in the message sequence.
798 1162
799=item * In Erlang, processes can be declared dead and later be found to be 1163If you want your software to be very reliable, you have to cope with
800alive. 1164corrupted and even out-of-order messages in both Erlang and AEMP. AEMP
801 1165simply tries to work better in common error cases, such as when a network
802In Erlang it can happen that a monitored process is declared dead and 1166link goes down.
803linked processes get killed, but later it turns out that the process is
804still alive - and can receive messages.
805
806In AEMP, when port monitoring detects a port as dead, then that port will
807eventually be killed - it cannot happen that a node detects a port as dead
808and then later sends messages to it, finding it is still alive.
809 1167
810=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.
811 1169
812In Erlang it is quite possible that a node that restarts reuses a process 1170In Erlang it is quite likely that a node that restarts reuses an Erlang
813ID known to other nodes for a completely different process, causing 1171process ID known to other nodes for a completely different process,
814messages destined for that process to end up in an unrelated process. 1172causing messages destined for that process to end up in an unrelated
1173process.
815 1174
816AEMP 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
817around in the network will not be sent to an unrelated port. 1176around in the network will not be sent to an unrelated port.
818 1177
819=item * Erlang uses unprotected connections, AEMP uses secure 1178=item * Erlang uses unprotected connections, AEMP uses secure
820authentication and can use TLS. 1179authentication and can use TLS.
821 1180
822AEMP can use a proven protocol - SSL/TLS - to protect connections and 1181AEMP can use a proven protocol - TLS - to protect connections and
823securely authenticate nodes. 1182securely authenticate nodes.
824 1183
825=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
826communications. 1185communications.
827 1186
828The AEMP protocol, unlike the Erlang protocol, supports both 1187The AEMP protocol, unlike the Erlang protocol, supports both programming
829language-independent text-only protocols (good for debugging) and binary, 1188language independent text-only protocols (good for debugging), and binary,
830language-specific serialisers (e.g. Storable). 1189language-specific serialisers (e.g. Storable). By default, unless TLS is
1190used, the protocol is actually completely text-based.
831 1191
832It has also been carefully designed to be implementable in other languages 1192It has also been carefully designed to be implementable in other languages
833with a minimum of work while gracefully degrading fucntionality to make the 1193with a minimum of work while gracefully degrading functionality to make the
834protocol simple. 1194protocol simple.
835 1195
836=item * AEMP has more flexible monitoring options than Erlang. 1196=item * AEMP has more flexible monitoring options than Erlang.
837 1197
838In 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
839or 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
840difficult to implement. Monitoring in AEMP is more flexible than in 1200difficult to implement.
841Erlang, as one can choose between automatic kill, exit message or callback 1201
842on 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.
843 1204
844=item * Erlang tries to hide remote/local connections, AEMP does not. 1205=item * Erlang tries to hide remote/local connections, AEMP does not.
845 1206
846Monitoring in Erlang is not an indicator of process death/crashes, 1207Monitoring in Erlang is not an indicator of process death/crashes, in the
847as linking is (except linking is unreliable in Erlang). 1208same way as linking is (except linking is unreliable in Erlang).
848 1209
849In AEMP, you don't "look up" registered port names or send to named ports 1210In AEMP, you don't "look up" registered port names or send to named ports
850that might or might not be persistent. Instead, you normally spawn a port 1211that might or might not be persistent. Instead, you normally spawn a port
851on the remote node. The init function monitors the you, and you monitor 1212on the remote node. The init function monitors you, and you monitor the
852the remote port. Since both monitors are local to the node, they are much 1213remote port. Since both monitors are local to the node, they are much more
853more reliable. 1214reliable (no need for C<spawn_link>).
854 1215
855This also saves round-trips and avoids sending messages to the wrong port 1216This also saves round-trips and avoids sending messages to the wrong port
856(hard to do in Erlang). 1217(hard to do in Erlang).
857 1218
858=back 1219=back
859 1220
860=head1 RATIONALE 1221=head1 RATIONALE
861 1222
862=over 4 1223=over 4
863 1224
864=item Why strings for ports and noderefs, why not objects? 1225=item Why strings for port and node IDs, why not objects?
865 1226
866We considered "objects", but found that the actual number of methods 1227We considered "objects", but found that the actual number of methods
867thatc an be called are very low. Since port IDs and noderefs travel over 1228that can be called are quite low. Since port and node IDs travel over
868the network frequently, the serialising/deserialising would add lots of 1229the network frequently, the serialising/deserialising would add lots of
869overhead, as well as having to keep a proxy object. 1230overhead, as well as having to keep a proxy object everywhere.
870 1231
871Strings can easily be printed, easily serialised etc. and need no special 1232Strings can easily be printed, easily serialised etc. and need no special
872procedures to be "valid". 1233procedures to be "valid".
873 1234
874And a a miniport consists of a single closure stored in a global hash - it 1235And as a result, a port with just a default receiver consists of a single
875can't become much cheaper. 1236code reference stored in a global hash - it can't become much cheaper.
876 1237
877=item Why favour JSON, why not real serialising format such as Storable? 1238=item Why favour JSON, why not a real serialising format such as Storable?
878 1239
879In fact, any AnyEvent::MP node will happily accept Storable as framing 1240In fact, any AnyEvent::MP node will happily accept Storable as framing
880format, 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
881default. 1242default (although all nodes will accept it).
882 1243
883The default framing protocol is JSON because a) JSON::XS is many times 1244The default framing protocol is JSON because a) JSON::XS is many times
884faster for small messages and b) most importantly, after years of 1245faster for small messages and b) most importantly, after years of
885experience we found that object serialisation is causing more problems 1246experience we found that object serialisation is causing more problems
886than it gains: Just like function calls, objects simply do not travel 1247than it solves: Just like function calls, objects simply do not travel
887easily over the network, mostly because they will always be a copy, so you 1248easily over the network, mostly because they will always be a copy, so you
888always have to re-think your design. 1249always have to re-think your design.
889 1250
890Keeping your messages simple, concentrating on data structures rather than 1251Keeping your messages simple, concentrating on data structures rather than
891objects, will keep your messages clean, tidy and efficient. 1252objects, will keep your messages clean, tidy and efficient.
892 1253
893=back 1254=back
894 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
895=head1 SEE ALSO 1364=head1 SEE ALSO
1365
1366L<AnyEvent::MP::Intro> - a gentle introduction.
1367
1368L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
1369
1370L<AnyEvent::MP::Global> - network maintenance and port groups, to find
1371your applications.
1372
1373L<AnyEvent::MP::DataConn> - establish data connections between nodes.
1374
1375L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
1376all nodes.
896 1377
897L<AnyEvent>. 1378L<AnyEvent>.
898 1379
899=head1 AUTHOR 1380=head1 AUTHOR
900 1381

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines