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.46 by root, Thu Aug 13 01:46:10 2009 UTC vs.
Revision 1.54 by root, Fri Aug 14 16:15:37 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
69 69
70=item port 70=item port
71 71
72A 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).
73 73
74Some 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
75messages. All C<rcv> handlers will receive messages they match, messages 75some messages. Messages will not be queued.
76will not be queued.
77 76
78=item port id - C<noderef#portname> 77=item port id - C<noderef#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 noderef, a hash-mark (C<#>) as
81separator, and a port name (a printable string of unspecified format). An 80separator, and a port name (a printable string of unspecified format). An
82exception 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
83reference. 82reference.
84 83
85=item node 84=item node
86 85
87A 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,
88port. 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
89create new ports, among other things. 88ports.
90 89
91Nodes are either private (single-process only), slaves (connected to a 90Nodes are either private (single-process only), slaves (connected to a
92master node only) or public nodes (connectable from unrelated nodes). 91master node only) or public nodes (connectable from unrelated nodes).
93 92
94=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>
143 kil $SELF, die => $msg; 142 kil $SELF, die => $msg;
144} 143}
145 144
146=item $thisnode = NODE / $NODE 145=item $thisnode = NODE / $NODE
147 146
148The C<NODE> function returns, and the C<$NODE> variable contains 147The C<NODE> function returns, and the C<$NODE> variable contains the
149the 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
150to C<become_public> or C<become_slave>, after which all local port 149C<initialise_node>.
151identifiers become invalid.
152 150
153=item $noderef = node_of $port 151=item $noderef = node_of $port
154 152
155Extracts and returns the noderef from a portid or a noderef. 153Extracts and returns the noderef from a port ID or a noderef.
156 154
157=item initialise_node $noderef, $seednode, $seednode... 155=item initialise_node $noderef, $seednode, $seednode...
158 156
159=item initialise_node "slave/", $master, $master... 157=item initialise_node "slave/", $master, $master...
160 158
163it should know the noderefs of some other nodes in the network. 161it should know the noderefs of some other nodes in the network.
164 162
165This function initialises a node - it must be called exactly once (or 163This function initialises a node - it must be called exactly once (or
166never) before calling other AnyEvent::MP functions. 164never) before calling other AnyEvent::MP functions.
167 165
168All 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.
169 174
170There are two types of networked nodes, public nodes and slave nodes: 175There are two types of networked nodes, public nodes and slave nodes:
171 176
172=over 4 177=over 4
173 178
174=item public nodes 179=item public nodes
175 180
176For public nodes, C<$noderef> must either be a (possibly unresolved) 181For public nodes, C<$noderef> (supplied either directly to
177noderef, 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
178which case the noderef will be guessed. 183noderef (possibly unresolved, in which case it will be resolved).
179 184
180Afterwards, 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
181to all additional C<$seednodes> that are specified. Seednodes are optional 186connect to all additional C<$seednodes> that are specified. Seednodes are
182and 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.
183 189
184=item slave nodes 190=item slave nodes
185 191
186When 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
187become 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
188route most of their traffic to the master node that they attach to. 195their traffic to the master node that they attach to.
189 196
190At least one additional noderef is required: The node will try to connect 197At least one additional noderef is required (either by specifying it
191to 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
192successfully 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.
193 201
194=back 202=back
195 203
196This 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
197nodes, until it has successfully established a connection to a master 205nodes, until it has successfully established a connection to a master
198server. 206server.
199 207
200Example: 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.
201 211
202 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";
203 223
204Example: 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
205servers to become part of the network. 225servers to become part of the network.
206 226
207 initialise_node undef, "master1", "master2"; 227 initialise_node undef, "master1", "master2";
210 230
211 initialise_node 4041; 231 initialise_node 4041;
212 232
213Example: become a public node, only visible on localhost port 4044. 233Example: become a public node, only visible on localhost port 4044.
214 234
215 initialise_node "locahost:4044"; 235 initialise_node "localhost: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 236
221=item $cv = resolve_node $noderef 237=item $cv = resolve_node $noderef
222 238
223Takes an unresolved node reference that may contain hostnames and 239Takes an unresolved node reference that may contain hostnames and
224abbreviated IDs, resolves all of them and returns a resolved node 240abbreviated IDs, resolves all of them and returns a resolved node
261=item snd $port, type => @data 277=item snd $port, type => @data
262 278
263=item snd $port, @msg 279=item snd $port, @msg
264 280
265Send 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
266a 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.
267stringifies a sa port ID (such as a port object :).
268 283
269While 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
270string 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
271type etc.). 286type etc.).
272 287
273The message data effectively becomes read-only after a call to this 288The message data effectively becomes read-only after a call to this
274function: modifying any argument is not allowed and can cause many 289function: modifying any argument is not allowed and can cause many
275problems. 290problems.
280that Storable can serialise and deserialise is allowed, and for the local 295that Storable can serialise and deserialise is allowed, and for the local
281node, anything can be passed. 296node, anything can be passed.
282 297
283=item $local_port = port 298=item $local_port = port
284 299
285Create 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
286matching port ("full port") or a single-callback port ("miniport"), 301no callbacks set and will throw an error when it receives messages.
287depending on how C<rcv> callbacks are bound to the object.
288 302
289=item $port = port { my @msg = @_; $finished } 303=item $local_port = port { my @msg = @_ }
290 304
291Creates a "miniport", that is, a very lightweight port without any pattern 305Creates 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. 306creating a port and calling C<rcv $port, $callback> on it.
294 307
295The 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
296callback 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
297will 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.
298 312
299The 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:
300be passed to the callback.
301 314
302If you need the local port id in the callback, this works nicely: 315 my $port = port {
303 316 my @msg = @_;
304 my $port; $port = port { 317 ...
305 snd $otherport, reply => $port; 318 kil $SELF;
306 }; 319 };
307 320
308=cut 321=cut
309 322
310sub rcv($@); 323sub rcv($@);
324
325sub _kilme {
326 die "received message on port without callback";
327}
311 328
312sub port(;&) { 329sub port(;&) {
313 my $id = "$UNIQ." . $ID++; 330 my $id = "$UNIQ." . $ID++;
314 my $port = "$NODE#$id"; 331 my $port = "$NODE#$id";
315 332
316 if (@_) { 333 rcv $port, shift || \&_kilme;
317 rcv $port, shift;
318 } else {
319 $PORT{$id} = sub { }; # nop
320 }
321 334
322 $port 335 $port
323} 336}
324 337
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) 338=item rcv $local_port, $callback->(@msg)
345 339
346Replaces the callback on the specified miniport (after converting it to 340Replaces the default callback on the specified port. There is no way to
347one if required). 341remove the default callback: use C<sub { }> to disable it, or better
348 342C<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 343
362The global C<$SELF> (exported by this module) contains C<$port> while 344The global C<$SELF> (exported by this module) contains C<$port> while
363executing the callback. 345executing the callback. Runtime errors during callback execution will
346result in the port being C<kil>ed.
364 347
365Runtime errors during callback execution will result in the port being 348The default callback received all messages not matched by a more specific
366C<kil>ed. 349C<tag> match.
367 350
368If the match is an array reference, then it will be matched against the 351=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
369first elements of the message, otherwise only the first element is being
370matched.
371 352
372Any element in the match that is specified as C<_any_> (a function 353Register (or replace) callbacks to be called on messages starting with the
373exported 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.
374 357
375While not required, it is highly recommended that the first matching 358The original message will be passed to the callback, after the first
376element is a string identifying the message. The one-string-only match is 359element (the tag) has been removed. The callback will use the same
377also the most efficient match (by far). 360environment as the default callback (see above).
378 361
379Example: create a port and bind receivers on it in one go. 362Example: create a port and bind receivers on it in one go.
380 363
381 my $port = rcv port, 364 my $port = rcv port,
382 msg1 => sub { ...; 0 }, 365 msg1 => sub { ... },
383 msg2 => sub { ...; 0 }, 366 msg2 => sub { ... },
384 ; 367 ;
385 368
386Example: 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
387in one go: 370in one go:
388 371
389 snd $otherport, reply => 372 snd $otherport, reply =>
390 rcv port, 373 rcv port,
391 msg1 => sub { ...; 0 }, 374 msg1 => sub { ... },
392 ... 375 ...
393 ; 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 };
394 386
395=cut 387=cut
396 388
397sub rcv($@) { 389sub rcv($@) {
398 my $port = shift; 390 my $port = shift;
399 my ($noderef, $portid) = split /#/, $port, 2; 391 my ($noderef, $portid) = split /#/, $port, 2;
400 392
401 ($NODE{$noderef} || add_node $noderef) == $NODE{""} 393 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
402 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";
403 395
404 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 {
405 my $cb = shift; 404 my $cb = shift;
406 delete $PORT_DATA{$portid};
407 $PORT{$portid} = sub { 405 $PORT{$portid} = sub {
408 local $SELF = $port; 406 local $SELF = $port;
409 eval { 407 eval { &$cb }; _self_die if $@;
410 &$cb 408 };
411 and kil $port;
412 }; 409 }
413 _self_die if $@; 410 } elsif (defined $_[0]) {
414 };
415 } else {
416 my $self = $PORT_DATA{$portid} ||= do { 411 my $self = $PORT_DATA{$portid} ||= do {
417 my $self = bless { 412 my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port";
418 id => $port,
419 }, "AnyEvent::MP::Port";
420 413
421 $PORT{$portid} = sub { 414 $PORT{$portid} = sub {
422 local $SELF = $port; 415 local $SELF = $port;
423 416
424 eval {
425 for (@{ $self->{rc0}{$_[0]} }) { 417 if (my $cb = $self->[1]{$_[0]}) {
426 $_ && &{$_->[0]} 418 shift;
427 && undef $_; 419 eval { &$cb }; _self_die if $@;
428 } 420 } else {
429
430 for (@{ $self->{rcv}{$_[0]} }) {
431 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
432 && &{$_->[0]} 421 &{ $self->[0] };
433 && undef $_;
434 }
435
436 for (@{ $self->{any} }) {
437 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
438 && &{$_->[0]}
439 && undef $_;
440 } 422 }
441 }; 423 };
442 _self_die if $@; 424
425 $self
443 }; 426 };
444 427
445 $self
446 };
447
448 "AnyEvent::MP::Port" eq ref $self 428 "AnyEvent::MP::Port" eq ref $self
449 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";
450 430
451 while (@_) {
452 my ($match, $cb) = splice @_, 0, 2; 431 my ($tag, $cb) = splice @_, 0, 2;
453 432
454 if (!ref $match) { 433 if (defined $cb) {
455 push @{ $self->{rc0}{$match} }, [$cb]; 434 $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 { 435 } else {
462 push @{ $self->{any} }, [$cb, $match]; 436 delete $self->[1]{$tag};
463 } 437 }
464 } 438 }
465 } 439 }
466 440
467 $port 441 $port
771convenience functionality. 745convenience functionality.
772 746
773This means that AEMP requires a less tightly controlled environment at the 747This means that AEMP requires a less tightly controlled environment at the
774cost of longer node references and a slightly higher management overhead. 748cost of longer node references and a slightly higher management overhead.
775 749
750=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
751uses "local ports are like remote ports".
752
753The failure modes for local ports are quite different (runtime errors
754only) then for remote ports - when a local port dies, you I<know> it dies,
755when a connection to another node dies, you know nothing about the other
756port.
757
758Erlang pretends remote ports are as reliable as local ports, even when
759they are not.
760
761AEMP encourages a "treat remote ports differently" philosophy, with local
762ports being the special case/exception, where transport errors cannot
763occur.
764
776=item * Erlang uses processes and a mailbox, AEMP does not queue. 765=item * Erlang uses processes and a mailbox, AEMP does not queue.
777 766
778Erlang uses processes that selctively receive messages, and therefore 767Erlang uses processes that selectively receive messages, and therefore
779needs a queue. AEMP is event based, queuing messages would serve no useful 768needs a queue. AEMP is event based, queuing messages would serve no
780purpose. 769useful purpose. For the same reason the pattern-matching abilities of
770AnyEvent::MP are more limited, as there is little need to be able to
771filter messages without dequeing them.
781 772
782(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 773(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
783 774
784=item * Erlang sends are synchronous, AEMP sends are asynchronous. 775=item * Erlang sends are synchronous, AEMP sends are asynchronous.
785 776
786Sending messages in Erlang is synchronous and blocks the process. AEMP 777Sending messages in Erlang is synchronous and blocks the process (and
787sends are immediate, connection establishment is handled in the 778so does not need a queue that can overflow). AEMP sends are immediate,
788background. 779connection establishment is handled in the background.
789 780
790=item * Erlang can silently lose messages, AEMP cannot. 781=item * Erlang suffers from silent message loss, AEMP does not.
791 782
792Erlang makes few guarantees on messages delivery - messages can get lost 783Erlang makes few guarantees on messages delivery - messages can get lost
793without any of the processes realising it (i.e. you send messages a, b, 784without any of the processes realising it (i.e. you send messages a, b,
794and c, and the other side only receives messages a and c). 785and c, and the other side only receives messages a and c).
795 786
807eventually be killed - it cannot happen that a node detects a port as dead 798eventually 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. 799and then later sends messages to it, finding it is still alive.
809 800
810=item * Erlang can send messages to the wrong port, AEMP does not. 801=item * Erlang can send messages to the wrong port, AEMP does not.
811 802
812In Erlang it is quite possible that a node that restarts reuses a process 803In Erlang it is quite likely that a node that restarts reuses a process ID
813ID known to other nodes for a completely different process, causing 804known to other nodes for a completely different process, causing messages
814messages destined for that process to end up in an unrelated process. 805destined for that process to end up in an unrelated process.
815 806
816AEMP never reuses port IDs, so old messages or old port IDs floating 807AEMP never reuses port IDs, so old messages or old port IDs floating
817around in the network will not be sent to an unrelated port. 808around in the network will not be sent to an unrelated port.
818 809
819=item * Erlang uses unprotected connections, AEMP uses secure 810=item * Erlang uses unprotected connections, AEMP uses secure
868the network frequently, the serialising/deserialising would add lots of 859the network frequently, the serialising/deserialising would add lots of
869overhead, as well as having to keep a proxy object. 860overhead, as well as having to keep a proxy object.
870 861
871Strings can easily be printed, easily serialised etc. and need no special 862Strings can easily be printed, easily serialised etc. and need no special
872procedures to be "valid". 863procedures to be "valid".
864
865And a a miniport consists of a single closure stored in a global hash - it
866can't become much cheaper.
873 867
874=item Why favour JSON, why not real serialising format such as Storable? 868=item Why favour JSON, why not real serialising format such as Storable?
875 869
876In fact, any AnyEvent::MP node will happily accept Storable as framing 870In fact, any AnyEvent::MP node will happily accept Storable as framing
877format, but currently there is no way to make a node use Storable by 871format, but currently there is no way to make a node use Storable by

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines