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.52 by root, Fri Aug 14 15:13:20 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 $somple_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
148 kil $SELF, die => $msg; 143 kil $SELF, die => $msg;
149} 144}
150 145
151=item $thisnode = NODE / $NODE 146=item $thisnode = NODE / $NODE
152 147
153The C<NODE> function returns, and the C<$NODE> variable contains 148The C<NODE> function returns, and the C<$NODE> variable contains the
154the noderef of the local node. The value is initialised by a call 149noderef 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 150C<initialise_node>.
156identifiers become invalid.
157 151
158=item $noderef = node_of $port 152=item $noderef = node_of $port
159 153
160Extracts and returns the noderef from a portid or a noderef. 154Extracts and returns the noderef from a port ID or a noderef.
161 155
162=item initialise_node $noderef, $seednode, $seednode... 156=item initialise_node $noderef, $seednode, $seednode...
163 157
164=item initialise_node "slave/", $master, $master... 158=item initialise_node "slave/", $master, $master...
165 159
168it should know the noderefs of some other nodes in the network. 162it should know the noderefs of some other nodes in the network.
169 163
170This function initialises a node - it must be called exactly once (or 164This function initialises a node - it must be called exactly once (or
171never) before calling other AnyEvent::MP functions. 165never) before calling other AnyEvent::MP functions.
172 166
173All arguments are noderefs, which can be either resolved or unresolved. 167All arguments (optionally except for the first) are noderefs, which can be
168either resolved or unresolved.
169
170The first argument will be looked up in the configuration database first
171(if it is C<undef> then the current nodename will be used instead) to find
172the relevant configuration profile (see L<aemp>). If none is found then
173the default configuration is used. The configuration supplies additional
174seed/master nodes and can override the actual noderef.
174 175
175There are two types of networked nodes, public nodes and slave nodes: 176There are two types of networked nodes, public nodes and slave nodes:
176 177
177=over 4 178=over 4
178 179
179=item public nodes 180=item public nodes
180 181
181For public nodes, C<$noderef> must either be a (possibly unresolved) 182For public nodes, C<$noderef> (supplied either directly to
182noderef, in which case it will be resolved, or C<undef> (or missing), in 183C<initialise_node> or indirectly via a profile or the nodename) must be a
183which case the noderef will be guessed. 184noderef (possibly unresolved, in which case it will be resolved).
184 185
185Afterwards, the node will bind itself on all endpoints and try to connect 186After resolving, the node will bind itself on all endpoints and try to
186to all additional C<$seednodes> that are specified. Seednodes are optional 187connect to all additional C<$seednodes> that are specified. Seednodes are
187and can be used to quickly bootstrap the node into an existing network. 188optional and can be used to quickly bootstrap the node into an existing
189network.
188 190
189=item slave nodes 191=item slave nodes
190 192
191When the C<$noderef> is the special string C<slave/>, then the node will 193When the C<$noderef> (either as given or overriden by the config file)
194is 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 195node. 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. 196their traffic to the master node that they attach to.
194 197
195At least one additional noderef is required: The node will try to connect 198At 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 199directly or because it is part of the configuration profile): The node
197successfully connect to. 200will try to connect to all of them and will become a slave attached to the
201first node it can successfully connect to.
198 202
199=back 203=back
200 204
201This function will block until all nodes have been resolved and, for slave 205This function will block until all nodes have been resolved and, for slave
202nodes, until it has successfully established a connection to a master 206nodes, until it has successfully established a connection to a master
203server. 207server.
204 208
205Example: become a public node listening on the default node. 209Example: become a public node listening on the guessed noderef, or the one
210specified via C<aemp> for the current node. This should be the most common
211form of invocation for "daemon"-type nodes.
206 212
207 initialise_node; 213 initialise_node;
214
215Example: become a slave node to any of the the seednodes specified via
216C<aemp>. This form is often used for commandline clients.
217
218 initialise_node "slave/";
219
220Example: become a slave node to any of the specified master servers. This
221form is also often used for commandline clients.
222
223 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
208 224
209Example: become a public node, and try to contact some well-known master 225Example: become a public node, and try to contact some well-known master
210servers to become part of the network. 226servers to become part of the network.
211 227
212 initialise_node undef, "master1", "master2"; 228 initialise_node undef, "master1", "master2";
215 231
216 initialise_node 4041; 232 initialise_node 4041;
217 233
218Example: become a public node, only visible on localhost port 4044. 234Example: become a public node, only visible on localhost port 4044.
219 235
220 initialise_node "locahost:4044"; 236 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 237
226=item $cv = resolve_node $noderef 238=item $cv = resolve_node $noderef
227 239
228Takes an unresolved node reference that may contain hostnames and 240Takes an unresolved node reference that may contain hostnames and
229abbreviated IDs, resolves all of them and returns a resolved node 241abbreviated IDs, resolves all of them and returns a resolved node
266=item snd $port, type => @data 278=item snd $port, type => @data
267 279
268=item snd $port, @msg 280=item snd $port, @msg
269 281
270Send the given message to the given port ID, which can identify either 282Send 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 283a local or a remote port, and must be a port ID.
272stringifies a sa port ID (such as a port object :).
273 284
274While the message can be about anything, it is highly recommended to use a 285While 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 286string as first element (a port ID, or some word that indicates a request
276type etc.). 287type etc.).
277 288
278The message data effectively becomes read-only after a call to this 289The message data effectively becomes read-only after a call to this
279function: modifying any argument is not allowed and can cause many 290function: modifying any argument is not allowed and can cause many
280problems. 291problems.
285that Storable can serialise and deserialise is allowed, and for the local 296that Storable can serialise and deserialise is allowed, and for the local
286node, anything can be passed. 297node, anything can be passed.
287 298
288=item $local_port = port 299=item $local_port = port
289 300
290Create a new local port object that can be used either as a pattern 301Create a new local port object and returns its port ID. Initially it has
291matching port ("full port") or a single-callback port ("miniport"), 302no callbacks set and will throw an error when it receives messages.
292depending on how C<rcv> callbacks are bound to the object.
293 303
294=item $port = port { my @msg = @_; $finished } 304=item $local_port = port { my @msg = @_ }
295 305
296Creates a "miniport", that is, a very lightweight port without any pattern 306Creates 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. 307creating a port and calling C<rcv $port, $callback> on it.
299 308
300The block will be called for every message received on the port. When the 309The 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 310global variable C<$SELF> set to the port ID. Runtime errors will cause the
302will be destroyed. Otherwise it will stay alive. 311port to be C<kil>ed. The message will be passed as-is, no extra argument
312(i.e. no port ID) will be passed to the callback.
303 313
304The message will be passed as-is, no extra argument (i.e. no port id) will 314If you want to stop/destroy the port, simply C<kil> it:
305be passed to the callback.
306 315
307If you need the local port id in the callback, this works nicely: 316 my $port = port {
308 317 my @msg = @_;
309 my $port; $port = port { 318 ...
310 snd $otherport, reply => $port; 319 kil $SELF;
311 }; 320 };
312 321
313=cut 322=cut
314 323
315sub rcv($@); 324sub rcv($@);
325
326sub _kilme {
327 die "received message on port without callback";
328}
316 329
317sub port(;&) { 330sub port(;&) {
318 my $id = "$UNIQ." . $ID++; 331 my $id = "$UNIQ." . $ID++;
319 my $port = "$NODE#$id"; 332 my $port = "$NODE#$id";
320 333
321 if (@_) { 334 rcv $port, shift || \&_kilme;
322 rcv $port, shift;
323 } else {
324 $PORT{$id} = sub { }; # nop
325 }
326 335
327 $port 336 $port
328} 337}
329 338
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) 339=item rcv $local_port, $callback->(@msg)
350 340
351Replaces the callback on the specified miniport (after converting it to 341Replaces the default callback on the specified port. There is no way to
352one if required). 342remove the default callback: use C<sub { }> to disable it, or better
353 343C<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 344
367The global C<$SELF> (exported by this module) contains C<$port> while 345The global C<$SELF> (exported by this module) contains C<$port> while
368executing the callback. 346executing the callback. Runtime errors during callback execution will
347result in the port being C<kil>ed.
369 348
370Runtime errors during callback execution will result in the port being 349The default callback received all messages not matched by a more specific
371C<kil>ed. 350C<tag> match.
372 351
373If the match is an array reference, then it will be matched against the 352=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
374first elements of the message, otherwise only the first element is being
375matched.
376 353
377Any element in the match that is specified as C<_any_> (a function 354Register callbacks to be called on messages starting with the given tag on
378exported by this module) matches any single element of the message. 355the given port (and return the port), or unregister it (when C<$callback>
356is C<$undef>).
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 ;
399 377
400=cut 378=cut
401 379
404 my ($noderef, $portid) = split /#/, $port, 2; 382 my ($noderef, $portid) = split /#/, $port, 2;
405 383
406 ($NODE{$noderef} || add_node $noderef) == $NODE{""} 384 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
407 or Carp::croak "$port: rcv can only be called on local ports, caught"; 385 or Carp::croak "$port: rcv can only be called on local ports, caught";
408 386
409 if (@_ == 1) { 387 while (@_) {
388 if (ref $_[0]) {
389 if (my $self = $PORT_DATA{$portid}) {
390 "AnyEvent::MP::Port" eq ref $self
391 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
392
393 $self->[2] = shift;
394 } else {
410 my $cb = shift; 395 my $cb = shift;
411 delete $PORT_DATA{$portid};
412 $PORT{$portid} = sub { 396 $PORT{$portid} = sub {
413 local $SELF = $port; 397 local $SELF = $port;
414 eval { 398 eval { &$cb }; _self_die if $@;
415 &$cb 399 };
416 and kil $port;
417 }; 400 }
418 _self_die if $@; 401 } elsif (defined $_[0]) {
419 };
420 } else {
421 my $self = $PORT_DATA{$portid} ||= do { 402 my $self = $PORT_DATA{$portid} ||= do {
422 my $self = bless { 403 my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port";
423 id => $port,
424 }, "AnyEvent::MP::Port";
425 404
426 $PORT{$portid} = sub { 405 $PORT{$portid} = sub {
427 local $SELF = $port; 406 local $SELF = $port;
428 407
429 eval {
430 for (@{ $self->{rc0}{$_[0]} }) { 408 if (my $cb = $self->[1]{$_[0]}) {
431 $_ && &{$_->[0]} 409 shift;
432 && undef $_; 410 eval { &$cb }; _self_die if $@;
433 } 411 } else {
434
435 for (@{ $self->{rcv}{$_[0]} }) {
436 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
437 && &{$_->[0]} 412 &{ $self->[0] };
438 && undef $_;
439 }
440
441 for (@{ $self->{any} }) {
442 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
443 && &{$_->[0]}
444 && undef $_;
445 } 413 }
446 }; 414 };
447 _self_die if $@; 415
416 $self
448 }; 417 };
449 418
450 $self
451 };
452
453 "AnyEvent::MP::Port" eq ref $self 419 "AnyEvent::MP::Port" eq ref $self
454 or Carp::croak "$port: rcv can only be called on message matching ports, caught"; 420 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
455 421
456 while (@_) {
457 my ($match, $cb) = splice @_, 0, 2; 422 my ($tag, $cb) = splice @_, 0, 2;
458 423
459 if (!ref $match) { 424 if (defined $cb) {
460 push @{ $self->{rc0}{$match} }, [$cb]; 425 $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 { 426 } else {
467 push @{ $self->{any} }, [$cb, $match]; 427 delete $self->[1]{$tag};
468 } 428 }
469 } 429 }
470 } 430 }
471 431
472 $port 432 $port
776convenience functionality. 736convenience functionality.
777 737
778This means that AEMP requires a less tightly controlled environment at the 738This means that AEMP requires a less tightly controlled environment at the
779cost of longer node references and a slightly higher management overhead. 739cost of longer node references and a slightly higher management overhead.
780 740
741=item Erlang has a "remote ports are like local ports" philosophy, AEMP
742uses "local ports are like remote ports".
743
744The failure modes for local ports are quite different (runtime errors
745only) then for remote ports - when a local port dies, you I<know> it dies,
746when a connection to another node dies, you know nothing about the other
747port.
748
749Erlang pretends remote ports are as reliable as local ports, even when
750they are not.
751
752AEMP encourages a "treat remote ports differently" philosophy, with local
753ports being the special case/exception, where transport errors cannot
754occur.
755
781=item * Erlang uses processes and a mailbox, AEMP does not queue. 756=item * Erlang uses processes and a mailbox, AEMP does not queue.
782 757
783Erlang uses processes that selctively receive messages, and therefore 758Erlang uses processes that selectively receive messages, and therefore
784needs a queue. AEMP is event based, queuing messages would serve no useful 759needs a queue. AEMP is event based, queuing messages would serve no
785purpose. 760useful purpose. For the same reason the pattern-matching abilities of
761AnyEvent::MP are more limited, as there is little need to be able to
762filter messages without dequeing them.
786 763
787(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP). 764(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
788 765
789=item * Erlang sends are synchronous, AEMP sends are asynchronous. 766=item * Erlang sends are synchronous, AEMP sends are asynchronous.
790 767
791Sending messages in Erlang is synchronous and blocks the process. AEMP 768Sending messages in Erlang is synchronous and blocks the process (and
792sends are immediate, connection establishment is handled in the 769so does not need a queue that can overflow). AEMP sends are immediate,
793background. 770connection establishment is handled in the background.
794 771
795=item * Erlang can silently lose messages, AEMP cannot. 772=item * Erlang suffers from silent message loss, AEMP does not.
796 773
797Erlang makes few guarantees on messages delivery - messages can get lost 774Erlang makes few guarantees on messages delivery - messages can get lost
798without any of the processes realising it (i.e. you send messages a, b, 775without 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). 776and c, and the other side only receives messages a and c).
800 777
812eventually be killed - it cannot happen that a node detects a port as dead 789eventually 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. 790and then later sends messages to it, finding it is still alive.
814 791
815=item * Erlang can send messages to the wrong port, AEMP does not. 792=item * Erlang can send messages to the wrong port, AEMP does not.
816 793
817In Erlang it is quite possible that a node that restarts reuses a process 794In 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 795known to other nodes for a completely different process, causing messages
819messages destined for that process to end up in an unrelated process. 796destined for that process to end up in an unrelated process.
820 797
821AEMP never reuses port IDs, so old messages or old port IDs floating 798AEMP never reuses port IDs, so old messages or old port IDs floating
822around in the network will not be sent to an unrelated port. 799around in the network will not be sent to an unrelated port.
823 800
824=item * Erlang uses unprotected connections, AEMP uses secure 801=item * Erlang uses unprotected connections, AEMP uses secure

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines