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.38 by root, Fri Aug 7 22:55:18 2009 UTC vs.
Revision 1.58 by root, Sun Aug 16 02:55:16 2009 UTC

9 $NODE # contains this node's noderef 9 $NODE # contains this node's noderef
10 NODE # returns this node's noderef 10 NODE # returns this node's noderef
11 NODE $port # returns the noderef of the port 11 NODE $port # returns the noderef of the port
12 12
13 $SELF # receiving/own port id in rcv callbacks 13 $SELF # receiving/own port id in rcv callbacks
14
15 # initialise the node so it can send/receive messages
16 initialise_node; # -OR-
17 initialise_node "localhost:4040"; # -OR-
18 initialise_node "slave/", "localhost:4040"
14 19
15 # ports are message endpoints 20 # ports are message endpoints
16 21
17 # sending messages 22 # sending messages
18 snd $port, type => data...; 23 snd $port, type => data...;
19 snd $port, @msg; 24 snd $port, @msg;
20 snd @msg_with_first_element_being_a_port; 25 snd @msg_with_first_element_being_a_port;
21 26
22 # miniports 27 # creating/using ports, the simple way
23 my $miniport = port { my @msg = @_; 0 }; 28 my $simple_port = port { my @msg = @_; 0 };
24 29
25 # full ports 30 # creating/using ports, tagged message matching
26 my $port = port; 31 my $port = port;
27 rcv $port, smartmatch => $cb->(@msg);
28 rcv $port, ping => sub { snd $_[0], "pong"; 0 }; 32 rcv $port, ping => sub { snd $_[0], "pong"; 0 };
29 rcv $port, pong => sub { warn "pong received\n"; 0 }; 33 rcv $port, pong => sub { warn "pong received\n"; 0 };
30 34
31 # remote ports 35 # create a port on another node
32 my $port = spawn $node, $initfunc, @initdata; 36 my $port = spawn $node, $initfunc, @initdata;
33
34 # more, smarter, matches (_any_ is exported by this module)
35 rcv $port, [child_died => $pid] => sub { ...
36 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3
37 37
38 # monitoring 38 # monitoring
39 mon $port, $cb->(@msg) # callback is invoked on death 39 mon $port, $cb->(@msg) # callback is invoked on death
40 mon $port, $otherport # kill otherport on abnormal death 40 mon $port, $otherport # kill otherport on abnormal death
41 mon $port, $otherport, @msg # send message on death 41 mon $port, $otherport, @msg # send message on death
42 42
43=head1 CURRENT STATUS
44
45 AnyEvent::MP - stable API, should work
46 AnyEvent::MP::Intro - outdated
47 AnyEvent::MP::Kernel - WIP
48 AnyEvent::MP::Transport - mostly stable
49
50 stay tuned.
51
43=head1 DESCRIPTION 52=head1 DESCRIPTION
44 53
45This module (-family) implements a simple message passing framework. 54This module (-family) implements a simple message passing framework.
46 55
47Despite its simplicity, you can securely message other processes running 56Despite its simplicity, you can securely message other processes running
50For 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>
51manual page. 60manual page.
52 61
53At the moment, this module family is severly broken and underdocumented, 62At the moment, this module family is severly broken and underdocumented,
54so do not use. This was uploaded mainly to reserve the CPAN namespace - 63so do not use. This was uploaded mainly to reserve the CPAN namespace -
55stay tuned! The basic API should be finished, however. 64stay tuned!
56 65
57=head1 CONCEPTS 66=head1 CONCEPTS
58 67
59=over 4 68=over 4
60 69
61=item port 70=item port
62 71
63A port is something you can send messages to (with the C<snd> function). 72A port is something you can send messages to (with the C<snd> function).
64 73
65Some ports allow you to register C<rcv> handlers that can match specific 74Ports allow you to register C<rcv> handlers that can match all or just
66messages. All C<rcv> handlers will receive messages they match, messages 75some messages. Messages will not be queued.
67will not be queued.
68 76
69=item port id - C<noderef#portname> 77=item port id - C<noderef#portname>
70 78
71A port id is normaly the concatenation of a noderef, a hash-mark (C<#>) as 79A port ID is the concatenation of a noderef, a hash-mark (C<#>) as
72separator, and a port name (a printable string of unspecified format). An 80separator, and a port name (a printable string of unspecified format). An
73exception is the the node port, whose ID is identical to its node 81exception is the the node port, whose ID is identical to its node
74reference. 82reference.
75 83
76=item node 84=item node
77 85
78A node is a single process containing at least one port - the node 86A node is a single process containing at least one port - the node port,
79port. You can send messages to node ports to find existing ports or to 87which provides nodes to manage each other remotely, and to create new
80create new ports, among other things. 88ports.
81 89
82Nodes are either private (single-process only), slaves (connected to a 90Nodes are either private (single-process only), slaves (connected to a
83master node only) or public nodes (connectable from unrelated nodes). 91master node only) or public nodes (connectable from unrelated nodes).
84 92
85=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id> 93=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
105 113
106=cut 114=cut
107 115
108package AnyEvent::MP; 116package AnyEvent::MP;
109 117
110use AnyEvent::MP::Base; 118use AnyEvent::MP::Kernel;
111 119
112use common::sense; 120use common::sense;
113 121
114use Carp (); 122use Carp ();
115 123
116use AE (); 124use AE ();
117 125
118use base "Exporter"; 126use base "Exporter";
119 127
120our $VERSION = '0.1'; 128our $VERSION = $AnyEvent::MP::Kernel::VERSION;
129
121our @EXPORT = qw( 130our @EXPORT = qw(
122 NODE $NODE *SELF node_of _any_ 131 NODE $NODE *SELF node_of _any_
123 resolve_node initialise_node 132 resolve_node initialise_node
124 snd rcv mon kil reg psub spawn 133 snd rcv mon kil reg psub spawn
125 port 134 port
133 kil $SELF, die => $msg; 142 kil $SELF, die => $msg;
134} 143}
135 144
136=item $thisnode = NODE / $NODE 145=item $thisnode = NODE / $NODE
137 146
138The C<NODE> function returns, and the C<$NODE> variable contains 147The C<NODE> function returns, and the C<$NODE> variable contains the
139the noderef of the local node. The value is initialised by a call 148noderef of the local node. The value is initialised by a call to
140to C<become_public> or C<become_slave>, after which all local port 149C<initialise_node>.
141identifiers become invalid.
142 150
143=item $noderef = node_of $port 151=item $noderef = node_of $port
144 152
145Extracts and returns the noderef from a portid or a noderef. 153Extracts and returns the noderef from a port ID or a noderef.
146 154
147=item initialise_node $noderef, $seednode, $seednode... 155=item initialise_node $noderef, $seednode, $seednode...
148 156
149=item initialise_node "slave/", $master, $master... 157=item initialise_node "slave/", $master, $master...
150 158
153it should know the noderefs of some other nodes in the network. 161it should know the noderefs of some other nodes in the network.
154 162
155This function initialises a node - it must be called exactly once (or 163This function initialises a node - it must be called exactly once (or
156never) before calling other AnyEvent::MP functions. 164never) before calling other AnyEvent::MP functions.
157 165
158All arguments are noderefs, which can be either resolved or unresolved. 166All arguments (optionally except for the first) are noderefs, which can be
167either resolved or unresolved.
168
169The first argument will be looked up in the configuration database first
170(if it is C<undef> then the current nodename will be used instead) to find
171the relevant configuration profile (see L<aemp>). If none is found then
172the default configuration is used. The configuration supplies additional
173seed/master nodes and can override the actual noderef.
159 174
160There are two types of networked nodes, public nodes and slave nodes: 175There are two types of networked nodes, public nodes and slave nodes:
161 176
162=over 4 177=over 4
163 178
164=item public nodes 179=item public nodes
165 180
166For public nodes, C<$noderef> must either be a (possibly unresolved) 181For public nodes, C<$noderef> (supplied either directly to
167noderef, in which case it will be resolved, or C<undef> (or missing), in 182C<initialise_node> or indirectly via a profile or the nodename) must be a
168which case the noderef will be guessed. 183noderef (possibly unresolved, in which case it will be resolved).
169 184
170Afterwards, the node will bind itself on all endpoints and try to connect 185After resolving, the node will bind itself on all endpoints and try to
171to all additional C<$seednodes> that are specified. Seednodes are optional 186connect to all additional C<$seednodes> that are specified. Seednodes are
172and can be used to quickly bootstrap the node into an existing network. 187optional and can be used to quickly bootstrap the node into an existing
188network.
173 189
174=item slave nodes 190=item slave nodes
175 191
176When the C<$noderef> is the special string C<slave/>, then the node will 192When the C<$noderef> (either as given or overriden by the config file)
193is the special string C<slave/>, then the node will become a slave
177become a slave node. Slave nodes cannot be contacted from outside and will 194node. Slave nodes cannot be contacted from outside and will route most of
178route most of their traffic to the master node that they attach to. 195their traffic to the master node that they attach to.
179 196
180At least one additional noderef is required: The node will try to connect 197At least one additional noderef is required (either by specifying it
181to all of them and will become a slave attached to the first node it can 198directly or because it is part of the configuration profile): The node
182successfully connect to. 199will try to connect to all of them and will become a slave attached to the
200first node it can successfully connect to.
201
202Note that slave nodes cannot change their name, and consequently, their
203master, so if the master goes down, the slave node will not function well
204anymore until it can re-establish conenciton to its master. This makes
205slave nodes unsuitable for long-term nodes or fault-tolerant networks.
183 206
184=back 207=back
185 208
186This function will block until all nodes have been resolved and, for slave 209This function will block until all nodes have been resolved and, for slave
187nodes, until it has successfully established a connection to a master 210nodes, until it has successfully established a connection to a master
188server. 211server.
189 212
213All the seednodes will also be specially marked to automatically retry
214connecting to them infinitely.
215
190Example: become a public node listening on the default node. 216Example: become a public node listening on the guessed noderef, or the one
217specified via C<aemp> for the current node. This should be the most common
218form of invocation for "daemon"-type nodes.
191 219
192 initialise_node; 220 initialise_node;
221
222Example: become a slave node to any of the the seednodes specified via
223C<aemp>. This form is often used for commandline clients.
224
225 initialise_node "slave/";
226
227Example: become a slave node to any of the specified master servers. This
228form is also often used for commandline clients.
229
230 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
193 231
194Example: become a public node, and try to contact some well-known master 232Example: become a public node, and try to contact some well-known master
195servers to become part of the network. 233servers to become part of the network.
196 234
197 initialise_node undef, "master1", "master2"; 235 initialise_node undef, "master1", "master2";
200 238
201 initialise_node 4041; 239 initialise_node 4041;
202 240
203Example: become a public node, only visible on localhost port 4044. 241Example: become a public node, only visible on localhost port 4044.
204 242
205 initialise_node "locahost:4044"; 243 initialise_node "localhost:4044";
206
207Example: become a slave node to any of the specified master servers.
208
209 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
210 244
211=item $cv = resolve_node $noderef 245=item $cv = resolve_node $noderef
212 246
213Takes an unresolved node reference that may contain hostnames and 247Takes an unresolved node reference that may contain hostnames and
214abbreviated IDs, resolves all of them and returns a resolved node 248abbreviated IDs, resolves all of them and returns a resolved node
251=item snd $port, type => @data 285=item snd $port, type => @data
252 286
253=item snd $port, @msg 287=item snd $port, @msg
254 288
255Send the given message to the given port ID, which can identify either 289Send the given message to the given port ID, which can identify either
256a local or a remote port, and can be either a string or soemthignt hat 290a local or a remote port, and must be a port ID.
257stringifies a sa port ID (such as a port object :).
258 291
259While the message can be about anything, it is highly recommended to use a 292While the message can be about anything, it is highly recommended to use a
260string as first element (a portid, or some word that indicates a request 293string as first element (a port ID, or some word that indicates a request
261type etc.). 294type etc.).
262 295
263The message data effectively becomes read-only after a call to this 296The message data effectively becomes read-only after a call to this
264function: modifying any argument is not allowed and can cause many 297function: modifying any argument is not allowed and can cause many
265problems. 298problems.
270that Storable can serialise and deserialise is allowed, and for the local 303that Storable can serialise and deserialise is allowed, and for the local
271node, anything can be passed. 304node, anything can be passed.
272 305
273=item $local_port = port 306=item $local_port = port
274 307
275Create a new local port object that can be used either as a pattern 308Create a new local port object and returns its port ID. Initially it has
276matching port ("full port") or a single-callback port ("miniport"), 309no callbacks set and will throw an error when it receives messages.
277depending on how C<rcv> callbacks are bound to the object.
278 310
279=item $port = port { my @msg = @_; $finished } 311=item $local_port = port { my @msg = @_ }
280 312
281Creates a "miniport", that is, a very lightweight port without any pattern 313Creates a new local port, and returns its ID. Semantically the same as
282matching behind it, and returns its ID. Semantically the same as creating
283a port and calling C<rcv $port, $callback> on it. 314creating a port and calling C<rcv $port, $callback> on it.
284 315
285The block will be called for every message received on the port. When the 316The block will be called for every message received on the port, with the
286callback returns a true value its job is considered "done" and the port 317global variable C<$SELF> set to the port ID. Runtime errors will cause the
287will be destroyed. Otherwise it will stay alive. 318port to be C<kil>ed. The message will be passed as-is, no extra argument
319(i.e. no port ID) will be passed to the callback.
288 320
289The message will be passed as-is, no extra argument (i.e. no port id) will 321If you want to stop/destroy the port, simply C<kil> it:
290be passed to the callback.
291 322
292If you need the local port id in the callback, this works nicely: 323 my $port = port {
293 324 my @msg = @_;
294 my $port; $port = port { 325 ...
295 snd $otherport, reply => $port; 326 kil $SELF;
296 }; 327 };
297 328
298=cut 329=cut
299 330
300sub rcv($@); 331sub rcv($@);
332
333sub _kilme {
334 die "received message on port without callback";
335}
301 336
302sub port(;&) { 337sub port(;&) {
303 my $id = "$UNIQ." . $ID++; 338 my $id = "$UNIQ." . $ID++;
304 my $port = "$NODE#$id"; 339 my $port = "$NODE#$id";
305 340
306 if (@_) { 341 rcv $port, shift || \&_kilme;
307 rcv $port, shift;
308 } else {
309 $PORT{$id} = sub { }; # nop
310 }
311 342
312 $port 343 $port
313} 344}
314 345
315=item reg $port, $name
316
317=item reg $name
318
319Registers the given port (or C<$SELF><<< if missing) under the name
320C<$name>. If the name already exists it is replaced.
321
322A port can only be registered under one well known name.
323
324A port automatically becomes unregistered when it is killed.
325
326=cut
327
328sub reg(@) {
329 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
330
331 $REG{$_[0]} = $port;
332}
333
334=item rcv $port, $callback->(@msg) 346=item rcv $local_port, $callback->(@msg)
335 347
336Replaces the callback on the specified miniport (after converting it to 348Replaces the default callback on the specified port. There is no way to
337one if required). 349remove the default callback: use C<sub { }> to disable it, or better
338 350C<kil> the port when it is no longer needed.
339=item rcv $port, tagstring => $callback->(@msg), ...
340
341=item rcv $port, $smartmatch => $callback->(@msg), ...
342
343=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
344
345Register callbacks to be called on matching messages on the given full
346port (after converting it to one if required) and return the port.
347
348The callback has to return a true value when its work is done, after
349which is will be removed, or a false value in which case it will stay
350registered.
351 351
352The global C<$SELF> (exported by this module) contains C<$port> while 352The global C<$SELF> (exported by this module) contains C<$port> while
353executing the callback. 353executing the callback. Runtime errors during callback execution will
354result in the port being C<kil>ed.
354 355
355Runtime errors during callback execution will result in the port being 356The default callback received all messages not matched by a more specific
356C<kil>ed. 357C<tag> match.
357 358
358If the match is an array reference, then it will be matched against the 359=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
359first elements of the message, otherwise only the first element is being
360matched.
361 360
362Any element in the match that is specified as C<_any_> (a function 361Register (or replace) callbacks to be called on messages starting with the
363exported by this module) matches any single element of the message. 362given tag on the given port (and return the port), or unregister it (when
363C<$callback> is C<$undef> or missing). There can only be one callback
364registered for each tag.
364 365
365While not required, it is highly recommended that the first matching 366The original message will be passed to the callback, after the first
366element is a string identifying the message. The one-string-only match is 367element (the tag) has been removed. The callback will use the same
367also the most efficient match (by far). 368environment as the default callback (see above).
368 369
369Example: create a port and bind receivers on it in one go. 370Example: create a port and bind receivers on it in one go.
370 371
371 my $port = rcv port, 372 my $port = rcv port,
372 msg1 => sub { ...; 0 }, 373 msg1 => sub { ... },
373 msg2 => sub { ...; 0 }, 374 msg2 => sub { ... },
374 ; 375 ;
375 376
376Example: create a port, bind receivers and send it in a message elsewhere 377Example: create a port, bind receivers and send it in a message elsewhere
377in one go: 378in one go:
378 379
379 snd $otherport, reply => 380 snd $otherport, reply =>
380 rcv port, 381 rcv port,
381 msg1 => sub { ...; 0 }, 382 msg1 => sub { ... },
382 ... 383 ...
383 ; 384 ;
385
386Example: temporarily register a rcv callback for a tag matching some port
387(e.g. for a rpc reply) and unregister it after a message was received.
388
389 rcv $port, $otherport => sub {
390 my @reply = @_;
391
392 rcv $SELF, $otherport;
393 };
384 394
385=cut 395=cut
386 396
387sub rcv($@) { 397sub rcv($@) {
388 my $port = shift; 398 my $port = shift;
389 my ($noderef, $portid) = split /#/, $port, 2; 399 my ($noderef, $portid) = split /#/, $port, 2;
390 400
391 ($NODE{$noderef} || add_node $noderef) == $NODE{""} 401 $NODE{$noderef} == $NODE{""}
392 or Carp::croak "$port: rcv can only be called on local ports, caught"; 402 or Carp::croak "$port: rcv can only be called on local ports, caught";
393 403
394 if (@_ == 1) { 404 while (@_) {
405 if (ref $_[0]) {
406 if (my $self = $PORT_DATA{$portid}) {
407 "AnyEvent::MP::Port" eq ref $self
408 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
409
410 $self->[2] = shift;
411 } else {
395 my $cb = shift; 412 my $cb = shift;
396 delete $PORT_DATA{$portid};
397 $PORT{$portid} = sub { 413 $PORT{$portid} = sub {
398 local $SELF = $port; 414 local $SELF = $port;
399 eval { 415 eval { &$cb }; _self_die if $@;
400 &$cb 416 };
401 and kil $port;
402 }; 417 }
403 _self_die if $@; 418 } elsif (defined $_[0]) {
404 };
405 } else {
406 my $self = $PORT_DATA{$portid} ||= do { 419 my $self = $PORT_DATA{$portid} ||= do {
407 my $self = bless { 420 my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port";
408 id => $port,
409 }, "AnyEvent::MP::Port";
410 421
411 $PORT{$portid} = sub { 422 $PORT{$portid} = sub {
412 local $SELF = $port; 423 local $SELF = $port;
413 424
414 eval {
415 for (@{ $self->{rc0}{$_[0]} }) { 425 if (my $cb = $self->[1]{$_[0]}) {
416 $_ && &{$_->[0]} 426 shift;
417 && undef $_; 427 eval { &$cb }; _self_die if $@;
418 } 428 } else {
419
420 for (@{ $self->{rcv}{$_[0]} }) {
421 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
422 && &{$_->[0]} 429 &{ $self->[0] };
423 && undef $_;
424 }
425
426 for (@{ $self->{any} }) {
427 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
428 && &{$_->[0]}
429 && undef $_;
430 } 430 }
431 }; 431 };
432 _self_die if $@; 432
433 $self
433 }; 434 };
434 435
435 $self
436 };
437
438 "AnyEvent::MP::Port" eq ref $self 436 "AnyEvent::MP::Port" eq ref $self
439 or Carp::croak "$port: rcv can only be called on message matching ports, caught"; 437 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
440 438
441 while (@_) {
442 my ($match, $cb) = splice @_, 0, 2; 439 my ($tag, $cb) = splice @_, 0, 2;
443 440
444 if (!ref $match) { 441 if (defined $cb) {
445 push @{ $self->{rc0}{$match} }, [$cb]; 442 $self->[1]{$tag} = $cb;
446 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
447 my ($type, @match) = @$match;
448 @match
449 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
450 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
451 } else { 443 } else {
452 push @{ $self->{any} }, [$cb, $match]; 444 delete $self->[1]{$tag};
453 } 445 }
454 } 446 }
455 } 447 }
456 448
457 $port 449 $port
501 493
502=item $guard = mon $port 494=item $guard = mon $port
503 495
504=item $guard = mon $port, $rcvport, @msg 496=item $guard = mon $port, $rcvport, @msg
505 497
506Monitor the given port and do something when the port is killed, and 498Monitor the given port and do something when the port is killed or
507optionally return a guard that can be used to stop monitoring again. 499messages to it were lost, and optionally return a guard that can be used
500to stop monitoring again.
501
502C<mon> effectively guarantees that, in the absence of hardware failures,
503that after starting the monitor, either all messages sent to the port
504will arrive, or the monitoring action will be invoked after possible
505message loss has been detected. No messages will be lost "in between"
506(after the first lost message no further messages will be received by the
507port). After the monitoring action was invoked, further messages might get
508delivered again.
509
510Note that monitoring-actions are one-shot: once released, they are removed
511and will not trigger again.
508 512
509In the first form (callback), the callback is simply called with any 513In the first form (callback), the callback is simply called with any
510number of C<@reason> elements (no @reason means that the port was deleted 514number of C<@reason> elements (no @reason means that the port was deleted
511"normally"). Note also that I<< the callback B<must> never die >>, so use 515"normally"). Note also that I<< the callback B<must> never die >>, so use
512C<eval> if unsure. 516C<eval> if unsure.
513 517
514In the second form (another port given), the other port (C<$rcvport) 518In the second form (another port given), the other port (C<$rcvport>)
515will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on 519will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on
516"normal" kils nothing happens, while under all other conditions, the other 520"normal" kils nothing happens, while under all other conditions, the other
517port is killed with the same reason. 521port is killed with the same reason.
518 522
519The third form (kill self) is the same as the second form, except that 523The third form (kill self) is the same as the second form, except that
546sub mon { 550sub mon {
547 my ($noderef, $port) = split /#/, shift, 2; 551 my ($noderef, $port) = split /#/, shift, 2;
548 552
549 my $node = $NODE{$noderef} || add_node $noderef; 553 my $node = $NODE{$noderef} || add_node $noderef;
550 554
551 my $cb = @_ ? $_[0] : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,'; 555 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
552 556
553 unless (ref $cb) { 557 unless (ref $cb) {
554 if (@_) { 558 if (@_) {
555 # send a kill info message 559 # send a kill info message
556 my (@msg) = @_; 560 my (@msg) = ($cb, @_);
557 $cb = sub { snd @msg, @_ }; 561 $cb = sub { snd @msg, @_ };
558 } else { 562 } else {
559 # simply kill other port 563 # simply kill other port
560 my $port = $cb; 564 my $port = $cb;
561 $cb = sub { kil $port, @_ if @_ }; 565 $cb = sub { kil $port, @_ if @_ };
619 623
620The port ID of the newly created port is return immediately, and it is 624The port ID of the newly created port is return immediately, and it is
621permissible to immediately start sending messages or monitor the port. 625permissible to immediately start sending messages or monitor the port.
622 626
623After the port has been created, the init function is 627After the port has been created, the init function is
624called. This fucntion must be a fully-qualified function name 628called. This function must be a fully-qualified function name
625(e.g. C<MyApp::Chat::Server::init>). 629(e.g. C<MyApp::Chat::Server::init>). To specify a function in the main
630program, use C<::name>.
626 631
627If the function doesn't exist, then the node tries to C<require> 632If the function doesn't exist, then the node tries to C<require>
628the package, then the package above the package and so on (e.g. 633the package, then the package above the package and so on (e.g.
629C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function 634C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
630exists or it runs out of package names. 635exists or it runs out of package names.
669sub spawn(@) { 674sub spawn(@) {
670 my ($noderef, undef) = split /#/, shift, 2; 675 my ($noderef, undef) = split /#/, shift, 2;
671 676
672 my $id = "$RUNIQ." . $ID++; 677 my $id = "$RUNIQ." . $ID++;
673 678
674 ($NODE{$noderef} || add_node $noderef) 679 $_[0] =~ /::/
675 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]); 680 or Carp::croak "spawn init function must be a fully-qualified name, caught";
681
682 snd_to_func $noderef, "AnyEvent::MP::_spawn" => $id, @_;
676 683
677 "$noderef#$id" 684 "$noderef#$id"
678} 685}
679
680=back
681
682=head1 NODE MESSAGES
683
684Nodes understand the following messages sent to them. Many of them take
685arguments called C<@reply>, which will simply be used to compose a reply
686message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
687the remaining arguments are simply the message data.
688
689While other messages exist, they are not public and subject to change.
690
691=over 4
692
693=cut
694
695=item lookup => $name, @reply
696
697Replies with the port ID of the specified well-known port, or C<undef>.
698
699=item devnull => ...
700
701Generic data sink/CPU heat conversion.
702
703=item relay => $port, @msg
704
705Simply forwards the message to the given port.
706
707=item eval => $string[ @reply]
708
709Evaluates the given string. If C<@reply> is given, then a message of the
710form C<@reply, $@, @evalres> is sent.
711
712Example: crash another node.
713
714 snd $othernode, eval => "exit";
715
716=item time => @reply
717
718Replies the the current node time to C<@reply>.
719
720Example: tell the current node to send the current time to C<$myport> in a
721C<timereply> message.
722
723 snd $NODE, time => $myport, timereply => 1, 2;
724 # => snd $myport, timereply => 1, 2, <time>
725 686
726=back 687=back
727 688
728=head1 AnyEvent::MP vs. Distributed Erlang 689=head1 AnyEvent::MP vs. Distributed Erlang
729 690
748convenience functionality. 709convenience functionality.
749 710
750This means that AEMP requires a less tightly controlled environment at the 711This means that AEMP requires a less tightly controlled environment at the
751cost of longer node references and a slightly higher management overhead. 712cost of longer node references and a slightly higher management overhead.
752 713
714=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
715uses "local ports are like remote ports".
716
717The failure modes for local ports are quite different (runtime errors
718only) then for remote ports - when a local port dies, you I<know> it dies,
719when a connection to another node dies, you know nothing about the other
720port.
721
722Erlang pretends remote ports are as reliable as local ports, even when
723they are not.
724
725AEMP encourages a "treat remote ports differently" philosophy, with local
726ports being the special case/exception, where transport errors cannot
727occur.
728
753=item * Erlang uses processes and a mailbox, AEMP does not queue. 729=item * Erlang uses processes and a mailbox, AEMP does not queue.
754 730
755Erlang uses processes that selctively receive messages, and therefore 731Erlang uses processes that selectively receive messages, and therefore
756needs a queue. AEMP is event based, queuing messages would serve no useful 732needs a queue. AEMP is event based, queuing messages would serve no
757purpose. 733useful purpose. For the same reason the pattern-matching abilities of
734AnyEvent::MP are more limited, as there is little need to be able to
735filter messages without dequeing them.
758 736
759(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 737(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
760 738
761=item * Erlang sends are synchronous, AEMP sends are asynchronous. 739=item * Erlang sends are synchronous, AEMP sends are asynchronous.
762 740
763Sending messages in Erlang is synchronous and blocks the process. AEMP 741Sending messages in Erlang is synchronous and blocks the process (and
764sends are immediate, connection establishment is handled in the 742so does not need a queue that can overflow). AEMP sends are immediate,
765background. 743connection establishment is handled in the background.
766 744
767=item * Erlang can silently lose messages, AEMP cannot. 745=item * Erlang suffers from silent message loss, AEMP does not.
768 746
769Erlang makes few guarantees on messages delivery - messages can get lost 747Erlang makes few guarantees on messages delivery - messages can get lost
770without any of the processes realising it (i.e. you send messages a, b, 748without any of the processes realising it (i.e. you send messages a, b,
771and c, and the other side only receives messages a and c). 749and c, and the other side only receives messages a and c).
772 750
784eventually be killed - it cannot happen that a node detects a port as dead 762eventually be killed - it cannot happen that a node detects a port as dead
785and then later sends messages to it, finding it is still alive. 763and then later sends messages to it, finding it is still alive.
786 764
787=item * Erlang can send messages to the wrong port, AEMP does not. 765=item * Erlang can send messages to the wrong port, AEMP does not.
788 766
789In Erlang it is quite possible that a node that restarts reuses a process 767In Erlang it is quite likely that a node that restarts reuses a process ID
790ID known to other nodes for a completely different process, causing 768known to other nodes for a completely different process, causing messages
791messages destined for that process to end up in an unrelated process. 769destined for that process to end up in an unrelated process.
792 770
793AEMP never reuses port IDs, so old messages or old port IDs floating 771AEMP never reuses port IDs, so old messages or old port IDs floating
794around in the network will not be sent to an unrelated port. 772around in the network will not be sent to an unrelated port.
795 773
796=item * Erlang uses unprotected connections, AEMP uses secure 774=item * Erlang uses unprotected connections, AEMP uses secure
832This also saves round-trips and avoids sending messages to the wrong port 810This also saves round-trips and avoids sending messages to the wrong port
833(hard to do in Erlang). 811(hard to do in Erlang).
834 812
835=back 813=back
836 814
815=head1 RATIONALE
816
817=over 4
818
819=item Why strings for ports and noderefs, why not objects?
820
821We considered "objects", but found that the actual number of methods
822thatc an be called are very low. Since port IDs and noderefs travel over
823the network frequently, the serialising/deserialising would add lots of
824overhead, as well as having to keep a proxy object.
825
826Strings can easily be printed, easily serialised etc. and need no special
827procedures to be "valid".
828
829And a a miniport consists of a single closure stored in a global hash - it
830can't become much cheaper.
831
832=item Why favour JSON, why not real serialising format such as Storable?
833
834In fact, any AnyEvent::MP node will happily accept Storable as framing
835format, but currently there is no way to make a node use Storable by
836default.
837
838The default framing protocol is JSON because a) JSON::XS is many times
839faster for small messages and b) most importantly, after years of
840experience we found that object serialisation is causing more problems
841than it gains: Just like function calls, objects simply do not travel
842easily over the network, mostly because they will always be a copy, so you
843always have to re-think your design.
844
845Keeping your messages simple, concentrating on data structures rather than
846objects, will keep your messages clean, tidy and efficient.
847
848=back
849
837=head1 SEE ALSO 850=head1 SEE ALSO
838 851
839L<AnyEvent>. 852L<AnyEvent>.
840 853
841=head1 AUTHOR 854=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines