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.48 by root, Thu Aug 13 02:59:42 2009 UTC vs.
Revision 1.55 by root, Fri Aug 14 23:17:17 2009 UTC

22 # sending messages 22 # sending messages
23 snd $port, type => data...; 23 snd $port, type => data...;
24 snd $port, @msg; 24 snd $port, @msg;
25 snd @msg_with_first_element_being_a_port; 25 snd @msg_with_first_element_being_a_port;
26 26
27 # creating/using miniports 27 # creating/using ports, the simple way
28 my $miniport = port { my @msg = @_; 0 }; 28 my $simple_port = port { my @msg = @_; 0 };
29 29
30 # creating/using full ports 30 # creating/using ports, tagged message matching
31 my $port = port; 31 my $port = port;
32 rcv $port, smartmatch => $cb->(@msg);
33 rcv $port, ping => sub { snd $_[0], "pong"; 0 }; 32 rcv $port, ping => sub { snd $_[0], "pong"; 0 };
34 rcv $port, pong => sub { warn "pong received\n"; 0 }; 33 rcv $port, pong => sub { warn "pong received\n"; 0 };
35
36 # more, smarter, matches (_any_ is exported by this module)
37 rcv $port, [child_died => $pid] => sub { ...
38 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3
39 34
40 # create a port on another node 35 # create a port on another node
41 my $port = spawn $node, $initfunc, @initdata; 36 my $port = spawn $node, $initfunc, @initdata;
42 37
43 # monitoring 38 # monitoring
74 69
75=item port 70=item port
76 71
77A 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).
78 73
79Some 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
80messages. All C<rcv> handlers will receive messages they match, messages 75some messages. Messages will not be queued.
81will not be queued.
82 76
83=item port id - C<noderef#portname> 77=item port id - C<noderef#portname>
84 78
85A 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
86separator, and a port name (a printable string of unspecified format). An 80separator, and a port name (a printable string of unspecified format). An
87exception 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
88reference. 82reference.
89 83
90=item node 84=item node
91 85
92A 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,
93port. 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
94create new ports, among other things. 88ports.
95 89
96Nodes are either private (single-process only), slaves (connected to a 90Nodes are either private (single-process only), slaves (connected to a
97master node only) or public nodes (connectable from unrelated nodes). 91master node only) or public nodes (connectable from unrelated nodes).
98 92
99=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>
148 kil $SELF, die => $msg; 142 kil $SELF, die => $msg;
149} 143}
150 144
151=item $thisnode = NODE / $NODE 145=item $thisnode = NODE / $NODE
152 146
153The C<NODE> function returns, and the C<$NODE> variable contains 147The C<NODE> function returns, and the C<$NODE> variable contains the
154the 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
155to C<become_public> or C<become_slave>, after which all local port 149C<initialise_node>.
156identifiers become invalid.
157 150
158=item $noderef = node_of $port 151=item $noderef = node_of $port
159 152
160Extracts and returns the noderef from a portid or a noderef. 153Extracts and returns the noderef from a port ID or a noderef.
161 154
162=item initialise_node $noderef, $seednode, $seednode... 155=item initialise_node $noderef, $seednode, $seednode...
163 156
164=item initialise_node "slave/", $master, $master... 157=item initialise_node "slave/", $master, $master...
165 158
168it should know the noderefs of some other nodes in the network. 161it should know the noderefs of some other nodes in the network.
169 162
170This function initialises a node - it must be called exactly once (or 163This function initialises a node - it must be called exactly once (or
171never) before calling other AnyEvent::MP functions. 164never) before calling other AnyEvent::MP functions.
172 165
173All 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.
174 174
175There are two types of networked nodes, public nodes and slave nodes: 175There are two types of networked nodes, public nodes and slave nodes:
176 176
177=over 4 177=over 4
178 178
179=item public nodes 179=item public nodes
180 180
181For public nodes, C<$noderef> must either be a (possibly unresolved) 181For public nodes, C<$noderef> (supplied either directly to
182noderef, 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
183which case the noderef will be guessed. 183noderef (possibly unresolved, in which case it will be resolved).
184 184
185Afterwards, 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
186to all additional C<$seednodes> that are specified. Seednodes are optional 186connect to all additional C<$seednodes> that are specified. Seednodes are
187and 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.
188 189
189=item slave nodes 190=item slave nodes
190 191
191When 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
192become 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
193route most of their traffic to the master node that they attach to. 195their traffic to the master node that they attach to.
194 196
195At least one additional noderef is required: The node will try to connect 197At least one additional noderef is required (either by specifying it
196to 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
197successfully 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.
198 201
199=back 202=back
200 203
201This function will block until all nodes have been resolved and, for slave 204This function will block until all nodes have been resolved and, for slave
202nodes, until it has successfully established a connection to a master 205nodes, until it has successfully established a connection to a master
203server. 206server.
204 207
205Example: become a public node listening on the default node. 208Example: become a public node listening on the guessed noderef, or the one
209specified via C<aemp> for the current node. This should be the most common
210form of invocation for "daemon"-type nodes.
206 211
207 initialise_node; 212 initialise_node;
213
214Example: become a slave node to any of the the seednodes specified via
215C<aemp>. This form is often used for commandline clients.
216
217 initialise_node "slave/";
218
219Example: become a slave node to any of the specified master servers. This
220form is also often used for commandline clients.
221
222 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
208 223
209Example: become a public node, and try to contact some well-known master 224Example: become a public node, and try to contact some well-known master
210servers to become part of the network. 225servers to become part of the network.
211 226
212 initialise_node undef, "master1", "master2"; 227 initialise_node undef, "master1", "master2";
215 230
216 initialise_node 4041; 231 initialise_node 4041;
217 232
218Example: become a public node, only visible on localhost port 4044. 233Example: become a public node, only visible on localhost port 4044.
219 234
220 initialise_node "locahost:4044"; 235 initialise_node "localhost:4044";
221
222Example: become a slave node to any of the specified master servers.
223
224 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
225 236
226=item $cv = resolve_node $noderef 237=item $cv = resolve_node $noderef
227 238
228Takes an unresolved node reference that may contain hostnames and 239Takes an unresolved node reference that may contain hostnames and
229abbreviated IDs, resolves all of them and returns a resolved node 240abbreviated IDs, resolves all of them and returns a resolved node
266=item snd $port, type => @data 277=item snd $port, type => @data
267 278
268=item snd $port, @msg 279=item snd $port, @msg
269 280
270Send the given message to the given port ID, which can identify either 281Send the given message to the given port ID, which can identify either
271a local or a remote port, and can be either a string or soemthignt hat 282a local or a remote port, and must be a port ID.
272stringifies a sa port ID (such as a port object :).
273 283
274While the message can be about anything, it is highly recommended to use a 284While the message can be about anything, it is highly recommended to use a
275string as first element (a portid, or some word that indicates a request 285string as first element (a port ID, or some word that indicates a request
276type etc.). 286type etc.).
277 287
278The message data effectively becomes read-only after a call to this 288The message data effectively becomes read-only after a call to this
279function: modifying any argument is not allowed and can cause many 289function: modifying any argument is not allowed and can cause many
280problems. 290problems.
285that Storable can serialise and deserialise is allowed, and for the local 295that Storable can serialise and deserialise is allowed, and for the local
286node, anything can be passed. 296node, anything can be passed.
287 297
288=item $local_port = port 298=item $local_port = port
289 299
290Create a new local port object that can be used either as a pattern 300Create a new local port object and returns its port ID. Initially it has
291matching port ("full port") or a single-callback port ("miniport"), 301no callbacks set and will throw an error when it receives messages.
292depending on how C<rcv> callbacks are bound to the object.
293 302
294=item $port = port { my @msg = @_; $finished } 303=item $local_port = port { my @msg = @_ }
295 304
296Creates a "miniport", that is, a very lightweight port without any pattern 305Creates a new local port, and returns its ID. Semantically the same as
297matching behind it, and returns its ID. Semantically the same as creating
298a port and calling C<rcv $port, $callback> on it. 306creating a port and calling C<rcv $port, $callback> on it.
299 307
300The block will be called for every message received on the port. When the 308The block will be called for every message received on the port, with the
301callback returns a true value its job is considered "done" and the port 309global variable C<$SELF> set to the port ID. Runtime errors will cause the
302will be destroyed. Otherwise it will stay alive. 310port to be C<kil>ed. The message will be passed as-is, no extra argument
311(i.e. no port ID) will be passed to the callback.
303 312
304The message will be passed as-is, no extra argument (i.e. no port id) will 313If you want to stop/destroy the port, simply C<kil> it:
305be passed to the callback.
306 314
307If you need the local port id in the callback, this works nicely: 315 my $port = port {
308 316 my @msg = @_;
309 my $port; $port = port { 317 ...
310 snd $otherport, reply => $port; 318 kil $SELF;
311 }; 319 };
312 320
313=cut 321=cut
314 322
315sub rcv($@); 323sub rcv($@);
324
325sub _kilme {
326 die "received message on port without callback";
327}
316 328
317sub port(;&) { 329sub port(;&) {
318 my $id = "$UNIQ." . $ID++; 330 my $id = "$UNIQ." . $ID++;
319 my $port = "$NODE#$id"; 331 my $port = "$NODE#$id";
320 332
321 if (@_) { 333 rcv $port, shift || \&_kilme;
322 rcv $port, shift;
323 } else {
324 $PORT{$id} = sub { }; # nop
325 }
326 334
327 $port 335 $port
328} 336}
329 337
330=item reg $port, $name
331
332=item reg $name
333
334Registers the given port (or C<$SELF><<< if missing) under the name
335C<$name>. If the name already exists it is replaced.
336
337A port can only be registered under one well known name.
338
339A port automatically becomes unregistered when it is killed.
340
341=cut
342
343sub reg(@) {
344 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
345
346 $REG{$_[0]} = $port;
347}
348
349=item rcv $port, $callback->(@msg) 338=item rcv $local_port, $callback->(@msg)
350 339
351Replaces the callback on the specified miniport (after converting it to 340Replaces the default callback on the specified port. There is no way to
352one if required). 341remove the default callback: use C<sub { }> to disable it, or better
353 342C<kil> the port when it is no longer needed.
354=item rcv $port, tagstring => $callback->(@msg), ...
355
356=item rcv $port, $smartmatch => $callback->(@msg), ...
357
358=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
359
360Register callbacks to be called on matching messages on the given full
361port (after converting it to one if required) and return the port.
362
363The callback has to return a true value when its work is done, after
364which is will be removed, or a false value in which case it will stay
365registered.
366 343
367The global C<$SELF> (exported by this module) contains C<$port> while 344The global C<$SELF> (exported by this module) contains C<$port> while
368executing the callback. 345executing the callback. Runtime errors during callback execution will
346result in the port being C<kil>ed.
369 347
370Runtime errors during callback execution will result in the port being 348The default callback received all messages not matched by a more specific
371C<kil>ed. 349C<tag> match.
372 350
373If the match is an array reference, then it will be matched against the 351=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
374first elements of the message, otherwise only the first element is being
375matched.
376 352
377Any element in the match that is specified as C<_any_> (a function 353Register (or replace) callbacks to be called on messages starting with the
378exported by this module) matches any single element of the message. 354given tag on the given port (and return the port), or unregister it (when
355C<$callback> is C<$undef> or missing). There can only be one callback
356registered for each tag.
379 357
380While not required, it is highly recommended that the first matching 358The original message will be passed to the callback, after the first
381element is a string identifying the message. The one-string-only match is 359element (the tag) has been removed. The callback will use the same
382also the most efficient match (by far). 360environment as the default callback (see above).
383 361
384Example: create a port and bind receivers on it in one go. 362Example: create a port and bind receivers on it in one go.
385 363
386 my $port = rcv port, 364 my $port = rcv port,
387 msg1 => sub { ...; 0 }, 365 msg1 => sub { ... },
388 msg2 => sub { ...; 0 }, 366 msg2 => sub { ... },
389 ; 367 ;
390 368
391Example: create a port, bind receivers and send it in a message elsewhere 369Example: create a port, bind receivers and send it in a message elsewhere
392in one go: 370in one go:
393 371
394 snd $otherport, reply => 372 snd $otherport, reply =>
395 rcv port, 373 rcv port,
396 msg1 => sub { ...; 0 }, 374 msg1 => sub { ... },
397 ... 375 ...
398 ; 376 ;
377
378Example: temporarily register a rcv callback for a tag matching some port
379(e.g. for a rpc reply) and unregister it after a message was received.
380
381 rcv $port, $otherport => sub {
382 my @reply = @_;
383
384 rcv $SELF, $otherport;
385 };
399 386
400=cut 387=cut
401 388
402sub rcv($@) { 389sub rcv($@) {
403 my $port = shift; 390 my $port = shift;
404 my ($noderef, $portid) = split /#/, $port, 2; 391 my ($noderef, $portid) = split /#/, $port, 2;
405 392
406 ($NODE{$noderef} || add_node $noderef) == $NODE{""} 393 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
407 or Carp::croak "$port: rcv can only be called on local ports, caught"; 394 or Carp::croak "$port: rcv can only be called on local ports, caught";
408 395
409 if (@_ == 1) { 396 while (@_) {
397 if (ref $_[0]) {
398 if (my $self = $PORT_DATA{$portid}) {
399 "AnyEvent::MP::Port" eq ref $self
400 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
401
402 $self->[2] = shift;
403 } else {
410 my $cb = shift; 404 my $cb = shift;
411 delete $PORT_DATA{$portid};
412 $PORT{$portid} = sub { 405 $PORT{$portid} = sub {
413 local $SELF = $port; 406 local $SELF = $port;
414 eval { 407 eval { &$cb }; _self_die if $@;
415 &$cb 408 };
416 and kil $port;
417 }; 409 }
418 _self_die if $@; 410 } elsif (defined $_[0]) {
419 };
420 } else {
421 my $self = $PORT_DATA{$portid} ||= do { 411 my $self = $PORT_DATA{$portid} ||= do {
422 my $self = bless { 412 my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port";
423 id => $port,
424 }, "AnyEvent::MP::Port";
425 413
426 $PORT{$portid} = sub { 414 $PORT{$portid} = sub {
427 local $SELF = $port; 415 local $SELF = $port;
428 416
429 eval {
430 for (@{ $self->{rc0}{$_[0]} }) { 417 if (my $cb = $self->[1]{$_[0]}) {
431 $_ && &{$_->[0]} 418 shift;
432 && undef $_; 419 eval { &$cb }; _self_die if $@;
433 } 420 } else {
434
435 for (@{ $self->{rcv}{$_[0]} }) {
436 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
437 && &{$_->[0]} 421 &{ $self->[0] };
438 && undef $_;
439 }
440
441 for (@{ $self->{any} }) {
442 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
443 && &{$_->[0]}
444 && undef $_;
445 } 422 }
446 }; 423 };
447 _self_die if $@; 424
425 $self
448 }; 426 };
449 427
450 $self
451 };
452
453 "AnyEvent::MP::Port" eq ref $self 428 "AnyEvent::MP::Port" eq ref $self
454 or Carp::croak "$port: rcv can only be called on message matching ports, caught"; 429 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
455 430
456 while (@_) {
457 my ($match, $cb) = splice @_, 0, 2; 431 my ($tag, $cb) = splice @_, 0, 2;
458 432
459 if (!ref $match) { 433 if (defined $cb) {
460 push @{ $self->{rc0}{$match} }, [$cb]; 434 $self->[1]{$tag} = $cb;
461 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
462 my ($type, @match) = @$match;
463 @match
464 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
465 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
466 } else { 435 } else {
467 push @{ $self->{any} }, [$cb, $match]; 436 delete $self->[1]{$tag};
468 } 437 }
469 } 438 }
470 } 439 }
471 440
472 $port 441 $port
697 my $id = "$RUNIQ." . $ID++; 666 my $id = "$RUNIQ." . $ID++;
698 667
699 $_[0] =~ /::/ 668 $_[0] =~ /::/
700 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 669 or Carp::croak "spawn init function must be a fully-qualified name, caught";
701 670
702 ($NODE{$noderef} || add_node $noderef) 671 snd_to_func $noderef, "AnyEvent::MP::_spawn" => $id, @_;
703 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
704 672
705 "$noderef#$id" 673 "$noderef#$id"
706} 674}
707 675
708=back 676=back
776convenience functionality. 744convenience functionality.
777 745
778This means that AEMP requires a less tightly controlled environment at the 746This means that AEMP requires a less tightly controlled environment at the
779cost of longer node references and a slightly higher management overhead. 747cost of longer node references and a slightly higher management overhead.
780 748
749=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
750uses "local ports are like remote ports".
751
752The failure modes for local ports are quite different (runtime errors
753only) then for remote ports - when a local port dies, you I<know> it dies,
754when a connection to another node dies, you know nothing about the other
755port.
756
757Erlang pretends remote ports are as reliable as local ports, even when
758they are not.
759
760AEMP encourages a "treat remote ports differently" philosophy, with local
761ports being the special case/exception, where transport errors cannot
762occur.
763
781=item * Erlang uses processes and a mailbox, AEMP does not queue. 764=item * Erlang uses processes and a mailbox, AEMP does not queue.
782 765
783Erlang uses processes that selctively receive messages, and therefore 766Erlang uses processes that selectively receive messages, and therefore
784needs a queue. AEMP is event based, queuing messages would serve no useful 767needs a queue. AEMP is event based, queuing messages would serve no
785purpose. 768useful purpose. For the same reason the pattern-matching abilities of
769AnyEvent::MP are more limited, as there is little need to be able to
770filter messages without dequeing them.
786 771
787(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 772(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
788 773
789=item * Erlang sends are synchronous, AEMP sends are asynchronous. 774=item * Erlang sends are synchronous, AEMP sends are asynchronous.
790 775
791Sending messages in Erlang is synchronous and blocks the process. AEMP 776Sending messages in Erlang is synchronous and blocks the process (and
792sends are immediate, connection establishment is handled in the 777so does not need a queue that can overflow). AEMP sends are immediate,
793background. 778connection establishment is handled in the background.
794 779
795=item * Erlang can silently lose messages, AEMP cannot. 780=item * Erlang suffers from silent message loss, AEMP does not.
796 781
797Erlang makes few guarantees on messages delivery - messages can get lost 782Erlang makes few guarantees on messages delivery - messages can get lost
798without any of the processes realising it (i.e. you send messages a, b, 783without any of the processes realising it (i.e. you send messages a, b,
799and c, and the other side only receives messages a and c). 784and c, and the other side only receives messages a and c).
800 785
812eventually be killed - it cannot happen that a node detects a port as dead 797eventually be killed - it cannot happen that a node detects a port as dead
813and then later sends messages to it, finding it is still alive. 798and then later sends messages to it, finding it is still alive.
814 799
815=item * Erlang can send messages to the wrong port, AEMP does not. 800=item * Erlang can send messages to the wrong port, AEMP does not.
816 801
817In Erlang it is quite possible that a node that restarts reuses a process 802In Erlang it is quite likely that a node that restarts reuses a process ID
818ID known to other nodes for a completely different process, causing 803known to other nodes for a completely different process, causing messages
819messages destined for that process to end up in an unrelated process. 804destined for that process to end up in an unrelated process.
820 805
821AEMP never reuses port IDs, so old messages or old port IDs floating 806AEMP never reuses port IDs, so old messages or old port IDs floating
822around in the network will not be sent to an unrelated port. 807around in the network will not be sent to an unrelated port.
823 808
824=item * Erlang uses unprotected connections, AEMP uses secure 809=item * Erlang uses unprotected connections, AEMP uses secure

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines