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.8 by root, Sun Aug 2 14:44:37 2009 UTC vs.
Revision 1.45 by root, Thu Aug 13 01:16:24 2009 UTC

4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::MP; 7 use AnyEvent::MP;
8 8
9 NODE # returns this node identifier
10 $NODE # contains this node identifier 9 $NODE # contains this node's noderef
10 NODE # returns this node's noderef
11 NODE $port # returns the noderef of the port
11 12
13 $SELF # receiving/own port id in rcv callbacks
14
15 # ports are message endpoints
16
17 # sending messages
12 snd $port, type => data...; 18 snd $port, type => data...;
19 snd $port, @msg;
20 snd @msg_with_first_element_being_a_port;
13 21
22 # miniports
23 my $miniport = port { my @msg = @_; 0 };
24
25 # full ports
26 my $port = port;
14 rcv $port, smartmatch => $cb->($port, @msg); 27 rcv $port, smartmatch => $cb->(@msg);
15
16 # examples:
17 rcv $port2, ping => sub { snd $_[0], "pong"; 0 }; 28 rcv $port, ping => sub { snd $_[0], "pong"; 0 };
18 rcv $port1, pong => sub { warn "pong received\n" }; 29 rcv $port, pong => sub { warn "pong received\n"; 0 };
19 snd $port2, ping => $port1; 30
31 # remote ports
32 my $port = spawn $node, $initfunc, @initdata;
20 33
21 # more, smarter, matches (_any_ is exported by this module) 34 # more, smarter, matches (_any_ is exported by this module)
22 rcv $port, [child_died => $pid] => sub { ... 35 rcv $port, [child_died => $pid] => sub { ...
23 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3 36 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3
24 37
38 # monitoring
39 mon $port, $cb->(@msg) # callback is invoked on death
40 mon $port, $otherport # kill otherport on abnormal death
41 mon $port, $otherport, @msg # send message on death
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
25=head1 DESCRIPTION 52=head1 DESCRIPTION
26 53
27This module (-family) implements a simple message passing framework. 54This module (-family) implements a simple message passing framework.
28 55
29Despite its simplicity, you can securely message other processes running 56Despite its simplicity, you can securely message other processes running
30on the same or other hosts. 57on the same or other hosts.
31 58
59For an introduction to this module family, see the L<AnyEvent::MP::Intro>
60manual page.
61
32At the moment, this module family is severly brokena nd underdocumented, 62At the moment, this module family is severly broken and underdocumented,
33so do not use. This was uploaded mainly to resreve the CPAN namespace - 63so do not use. This was uploaded mainly to reserve the CPAN namespace -
34stay tuned! 64stay tuned!
35 65
36=head1 CONCEPTS 66=head1 CONCEPTS
37 67
38=over 4 68=over 4
39 69
40=item port 70=item port
41 71
42A port is something you can send messages to with the C<snd> function, and 72A port is something you can send messages to (with the C<snd> function).
43you can register C<rcv> handlers with. All C<rcv> handlers will receive 73
44messages they match, messages will not be queued. 74Some ports allow you to register C<rcv> handlers that can match specific
75messages. All C<rcv> handlers will receive messages they match, messages
76will not be queued.
45 77
46=item port id - C<noderef#portname> 78=item port id - C<noderef#portname>
47 79
48A port id is always the noderef, a hash-mark (C<#>) as separator, followed 80A port id is normaly the concatenation of a noderef, a hash-mark (C<#>) as
49by a port name (a printable string of unspecified format). 81separator, and a port name (a printable string of unspecified format). An
82exception is the the node port, whose ID is identical to its node
83reference.
50 84
51=item node 85=item node
52 86
53A node is a single process containing at least one port - the node 87A node is a single process containing at least one port - the node
54port. You can send messages to node ports to let them create new ports, 88port. You can send messages to node ports to find existing ports or to
55among other things. 89create new ports, among other things.
56 90
57Initially, nodes are either private (single-process only) or hidden 91Nodes are either private (single-process only), slaves (connected to a
58(connected to a master node only). Only when they epxlicitly "become 92master node only) or public nodes (connectable from unrelated nodes).
59public" can you send them messages from unrelated other nodes.
60 93
61=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id> 94=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
62 95
63A noderef is a string that either uniquely identifies a given node (for 96A node reference is a string that either simply identifies the node (for
64private and hidden nodes), or contains a recipe on how to reach a given 97private and slave nodes), or contains a recipe on how to reach a given
65node (for public nodes). 98node (for public nodes).
66 99
100This recipe is simply a comma-separated list of C<address:port> pairs (for
101TCP/IP, other protocols might look different).
102
103Node references come in two flavours: resolved (containing only numerical
104addresses) or unresolved (where hostnames are used instead of addresses).
105
106Before using an unresolved node reference in a message you first have to
107resolve it.
108
67=back 109=back
68 110
69=head1 VARIABLES/FUNCTIONS 111=head1 VARIABLES/FUNCTIONS
70 112
71=over 4 113=over 4
72 114
73=cut 115=cut
74 116
75package AnyEvent::MP; 117package AnyEvent::MP;
76 118
77use AnyEvent::MP::Base; 119use AnyEvent::MP::Kernel;
78 120
79use common::sense; 121use common::sense;
80 122
81use Carp (); 123use Carp ();
82 124
83use AE (); 125use AE ();
84 126
85use base "Exporter"; 127use base "Exporter";
86 128
87our $VERSION = '0.01'; 129our $VERSION = $AnyEvent::MP::Kernel::VERSION;
130
88our @EXPORT = qw( 131our @EXPORT = qw(
89 NODE $NODE $PORT snd rcv _any_ 132 NODE $NODE *SELF node_of _any_
90 create_port create_port_on 133 resolve_node initialise_node
91 become_slave become_public 134 snd rcv mon kil reg psub spawn
135 port
92); 136);
93 137
138our $SELF;
139
140sub _self_die() {
141 my $msg = $@;
142 $msg =~ s/\n+$// unless ref $msg;
143 kil $SELF, die => $msg;
144}
145
94=item NODE / $NODE 146=item $thisnode = NODE / $NODE
95 147
96The C<NODE ()> function and the C<$NODE> variable contain the noderef of 148The C<NODE> function returns, and the C<$NODE> variable contains
97the local node. The value is initialised by a call to C<become_public> or 149the noderef of the local node. The value is initialised by a call
98C<become_slave>, after which all local port identifiers become invalid. 150to C<become_public> or C<become_slave>, after which all local port
151identifiers become invalid.
99 152
153=item $noderef = node_of $port
154
155Extracts and returns the noderef from a portid or a noderef.
156
157=item initialise_node $noderef, $seednode, $seednode...
158
159=item initialise_node "slave/", $master, $master...
160
161Before a node can talk to other nodes on the network it has to initialise
162itself - the minimum a node needs to know is it's own name, and optionally
163it should know the noderefs of some other nodes in the network.
164
165This function initialises a node - it must be called exactly once (or
166never) before calling other AnyEvent::MP functions.
167
168All arguments are noderefs, which can be either resolved or unresolved.
169
170There are two types of networked nodes, public nodes and slave nodes:
171
172=over 4
173
174=item public nodes
175
176For public nodes, C<$noderef> must either be a (possibly unresolved)
177noderef, in which case it will be resolved, or C<undef> (or missing), in
178which case the noderef will be guessed.
179
180Afterwards, the node will bind itself on all endpoints and try to connect
181to all additional C<$seednodes> that are specified. Seednodes are optional
182and can be used to quickly bootstrap the node into an existing network.
183
184=item slave nodes
185
186When the C<$noderef> is the special string C<slave/>, then the node will
187become a slave node. Slave nodes cannot be contacted from outside and will
188route most of their traffic to the master node that they attach to.
189
190At least one additional noderef is required: The node will try to connect
191to all of them and will become a slave attached to the first node it can
192successfully connect to.
193
194=back
195
196This function will block until all nodes have been resolved and, for slave
197nodes, until it has successfully established a connection to a master
198server.
199
200Example: become a public node listening on the default node.
201
202 initialise_node;
203
204Example: become a public node, and try to contact some well-known master
205servers to become part of the network.
206
207 initialise_node undef, "master1", "master2";
208
209Example: become a public node listening on port C<4041>.
210
211 initialise_node 4041;
212
213Example: become a public node, only visible on localhost port 4044.
214
215 initialise_node "locahost:4044";
216
217Example: become a slave node to any of the specified master servers.
218
219 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
220
221=item $cv = resolve_node $noderef
222
223Takes an unresolved node reference that may contain hostnames and
224abbreviated IDs, resolves all of them and returns a resolved node
225reference.
226
227In addition to C<address:port> pairs allowed in resolved noderefs, the
228following forms are supported:
229
230=over 4
231
232=item the empty string
233
234An empty-string component gets resolved as if the default port (4040) was
235specified.
236
237=item naked port numbers (e.g. C<1234>)
238
239These are resolved by prepending the local nodename and a colon, to be
240further resolved.
241
242=item hostnames (e.g. C<localhost:1234>, C<localhost>)
243
244These are resolved by using AnyEvent::DNS to resolve them, optionally
245looking up SRV records for the C<aemp=4040> port, if no port was
246specified.
247
248=back
249
250=item $SELF
251
252Contains the current port id while executing C<rcv> callbacks or C<psub>
253blocks.
254
255=item SELF, %SELF, @SELF...
256
257Due to some quirks in how perl exports variables, it is impossible to
258just export C<$SELF>, all the symbols called C<SELF> are exported by this
259module, but only C<$SELF> is currently used.
260
100=item snd $portid, type => @data 261=item snd $port, type => @data
101 262
102=item snd $portid, @msg 263=item snd $port, @msg
103 264
104Send the given message to the given port ID, which can identify either 265Send the given message to the given port ID, which can identify either
105a local or a remote port, and can be either a string or soemthignt hat 266a local or a remote port, and can be either a string or soemthignt hat
106stringifies a sa port ID (such as a port object :). 267stringifies a sa port ID (such as a port object :).
107 268
117JSON is used, then only strings, numbers and arrays and hashes consisting 278JSON is used, then only strings, numbers and arrays and hashes consisting
118of those are allowed (no objects). When Storable is used, then anything 279of those are allowed (no objects). When Storable is used, then anything
119that Storable can serialise and deserialise is allowed, and for the local 280that Storable can serialise and deserialise is allowed, and for the local
120node, anything can be passed. 281node, anything can be passed.
121 282
122=item $local_port = create_port 283=item $local_port = port
123 284
124Create a new local port object. See the next section for allowed methods. 285Create a new local port object that can be used either as a pattern
286matching port ("full port") or a single-callback port ("miniport"),
287depending on how C<rcv> callbacks are bound to the object.
125 288
126=cut 289=item $port = port { my @msg = @_; $finished }
127 290
128sub create_port { 291Creates a "miniport", that is, a very lightweight port without any pattern
129 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID; 292matching behind it, and returns its ID. Semantically the same as creating
293a port and calling C<rcv $port, $callback> on it.
130 294
131 my $self = bless { 295The block will be called for every message received on the port. When the
132 id => "$NODE#$id", 296callback returns a true value its job is considered "done" and the port
133 names => [$id], 297will be destroyed. Otherwise it will stay alive.
134 }, "AnyEvent::MP::Port";
135 298
136 $AnyEvent::MP::Base::PORT{$id} = sub { 299The message will be passed as-is, no extra argument (i.e. no port id) will
137 unshift @_, $self; 300be passed to the callback.
138 301
139 for (@{ $self->{rc0}{$_[1]} }) { 302If you need the local port id in the callback, this works nicely:
140 $_ && &{$_->[0]}
141 && undef $_;
142 }
143 303
144 for (@{ $self->{rcv}{$_[1]} }) { 304 my $port; $port = port {
145 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1] 305 snd $otherport, reply => $port;
146 && &{$_->[0]}
147 && undef $_;
148 }
149
150 for (@{ $self->{any} }) {
151 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
152 && &{$_->[0]}
153 && undef $_;
154 }
155 }; 306 };
156 307
157 $self 308=cut
309
310sub rcv($@);
311
312sub port(;&) {
313 my $id = "$UNIQ." . $ID++;
314 my $port = "$NODE#$id";
315
316 if (@_) {
317 rcv $port, shift;
318 } else {
319 $PORT{$id} = sub { }; # nop
320 }
321
322 $port
158} 323}
159 324
160package AnyEvent::MP::Port; 325=item reg $port, $name
161 326
162=back 327=item reg $name
163 328
164=head1 METHODS FOR PORT OBJECTS 329Registers the given port (or C<$SELF><<< if missing) under the name
330C<$name>. If the name already exists it is replaced.
165 331
166=over 4 332A port can only be registered under one well known name.
167 333
168=item "$port" 334A port automatically becomes unregistered when it is killed.
169 335
170A port object stringifies to its port ID, so can be used directly for
171C<snd> operations.
172
173=cut 336=cut
174 337
175use overload 338sub reg(@) {
176 '""' => sub { $_[0]{id} }, 339 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
177 fallback => 1;
178 340
179=item $port->rcv (type => $callback->($port, @msg)) 341 $REG{$_[0]} = $port;
342}
180 343
181=item $port->rcv ($smartmatch => $callback->($port, @msg)) 344=item rcv $port, $callback->(@msg)
182 345
346Replaces the callback on the specified miniport (after converting it to
347one if required).
348
349=item rcv $port, tagstring => $callback->(@msg), ...
350
351=item rcv $port, $smartmatch => $callback->(@msg), ...
352
183=item $port->rcv ([$smartmatch...] => $callback->($port, @msg)) 353=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
184 354
185Register a callback on the given port. 355Register callbacks to be called on matching messages on the given full
356port (after converting it to one if required) and return the port.
186 357
187The callback has to return a true value when its work is done, after 358The callback has to return a true value when its work is done, after
188which is will be removed, or a false value in which case it will stay 359which is will be removed, or a false value in which case it will stay
189registered. 360registered.
190 361
362The global C<$SELF> (exported by this module) contains C<$port> while
363executing the callback.
364
365Runtime errors during callback execution will result in the port being
366C<kil>ed.
367
191If the match is an array reference, then it will be matched against the 368If the match is an array reference, then it will be matched against the
192first elements of the message, otherwise only the first element is being 369first elements of the message, otherwise only the first element is being
193matched. 370matched.
194 371
195Any element in the match that is specified as C<_any_> (a function 372Any element in the match that is specified as C<_any_> (a function
197 374
198While not required, it is highly recommended that the first matching 375While not required, it is highly recommended that the first matching
199element is a string identifying the message. The one-string-only match is 376element is a string identifying the message. The one-string-only match is
200also the most efficient match (by far). 377also the most efficient match (by far).
201 378
379Example: create a port and bind receivers on it in one go.
380
381 my $port = rcv port,
382 msg1 => sub { ...; 0 },
383 msg2 => sub { ...; 0 },
384 ;
385
386Example: create a port, bind receivers and send it in a message elsewhere
387in one go:
388
389 snd $otherport, reply =>
390 rcv port,
391 msg1 => sub { ...; 0 },
392 ...
393 ;
394
202=cut 395=cut
203 396
204sub rcv($@) { 397sub rcv($@) {
205 my ($self, $match, $cb) = @_; 398 my $port = shift;
399 my ($noderef, $portid) = split /#/, $port, 2;
206 400
207 if (!ref $match) { 401 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
208 push @{ $self->{rc0}{$match} }, [$cb]; 402 or Carp::croak "$port: rcv can only be called on local ports, caught";
209 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 403
210 my ($type, @match) = @$match; 404 if (@_ == 1) {
211 @match 405 my $cb = shift;
212 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match] 406 delete $PORT_DATA{$portid};
213 : push @{ $self->{rc0}{$match->[0]} }, [$cb]; 407 $PORT{$portid} = sub {
408 local $SELF = $port;
409 eval {
410 &$cb
411 and kil $port;
412 };
413 _self_die if $@;
414 };
214 } else { 415 } else {
416 my $self = $PORT_DATA{$portid} ||= do {
417 my $self = bless {
418 id => $port,
419 }, "AnyEvent::MP::Port";
420
421 $PORT{$portid} = sub {
422 local $SELF = $port;
423
424 eval {
425 for (@{ $self->{rc0}{$_[0]} }) {
426 $_ && &{$_->[0]}
427 && undef $_;
428 }
429
430 for (@{ $self->{rcv}{$_[0]} }) {
431 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
432 && &{$_->[0]}
433 && undef $_;
434 }
435
436 for (@{ $self->{any} }) {
437 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
438 && &{$_->[0]}
439 && undef $_;
440 }
441 };
442 _self_die if $@;
443 };
444
445 $self
446 };
447
448 "AnyEvent::MP::Port" eq ref $self
449 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
450
451 while (@_) {
452 my ($match, $cb) = splice @_, 0, 2;
453
454 if (!ref $match) {
455 push @{ $self->{rc0}{$match} }, [$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 {
215 push @{ $self->{any} }, [$cb, $match]; 462 push @{ $self->{any} }, [$cb, $match];
463 }
464 }
465 }
466
467 $port
468}
469
470=item $closure = psub { BLOCK }
471
472Remembers C<$SELF> and creates a closure out of the BLOCK. When the
473closure is executed, sets up the environment in the same way as in C<rcv>
474callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
475
476This is useful when you register callbacks from C<rcv> callbacks:
477
478 rcv delayed_reply => sub {
479 my ($delay, @reply) = @_;
480 my $timer = AE::timer $delay, 0, psub {
481 snd @reply, $SELF;
482 };
483 };
484
485=cut
486
487sub psub(&) {
488 my $cb = shift;
489
490 my $port = $SELF
491 or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
492
493 sub {
494 local $SELF = $port;
495
496 if (wantarray) {
497 my @res = eval { &$cb };
498 _self_die if $@;
499 @res
500 } else {
501 my $res = eval { &$cb };
502 _self_die if $@;
503 $res
504 }
216 } 505 }
217} 506}
218 507
219=item $port->register ($name) 508=item $guard = mon $port, $cb->(@reason)
220 509
221Registers the given port under the well known name C<$name>. If the name 510=item $guard = mon $port, $rcvport
222already exists it is replaced.
223 511
224A port can only be registered under one well known name. 512=item $guard = mon $port
225 513
226=cut 514=item $guard = mon $port, $rcvport, @msg
227 515
228sub register { 516Monitor the given port and do something when the port is killed or
229 my ($self, $name) = @_; 517messages to it were lost, and optionally return a guard that can be used
518to stop monitoring again.
230 519
231 $self->{wkname} = $name; 520C<mon> effectively guarantees that, in the absence of hardware failures,
232 $AnyEvent::MP::Base::WKP{$name} = "$self"; 521that after starting the monitor, either all messages sent to the port
522will arrive, or the monitoring action will be invoked after possible
523message loss has been detected. No messages will be lost "in between"
524(after the first lost message no further messages will be received by the
525port). After the monitoring action was invoked, further messages might get
526delivered again.
527
528In the first form (callback), the callback is simply called with any
529number of C<@reason> elements (no @reason means that the port was deleted
530"normally"). Note also that I<< the callback B<must> never die >>, so use
531C<eval> if unsure.
532
533In the second form (another port given), the other port (C<$rcvport>)
534will be C<kil>'ed with C<@reason>, iff a @reason was specified, i.e. on
535"normal" kils nothing happens, while under all other conditions, the other
536port is killed with the same reason.
537
538The third form (kill self) is the same as the second form, except that
539C<$rvport> defaults to C<$SELF>.
540
541In the last form (message), a message of the form C<@msg, @reason> will be
542C<snd>.
543
544As a rule of thumb, monitoring requests should always monitor a port from
545a local port (or callback). The reason is that kill messages might get
546lost, just like any other message. Another less obvious reason is that
547even monitoring requests can get lost (for exmaple, when the connection
548to the other node goes down permanently). When monitoring a port locally
549these problems do not exist.
550
551Example: call a given callback when C<$port> is killed.
552
553 mon $port, sub { warn "port died because of <@_>\n" };
554
555Example: kill ourselves when C<$port> is killed abnormally.
556
557 mon $port;
558
559Example: send us a restart message when another C<$port> is killed.
560
561 mon $port, $self => "restart";
562
563=cut
564
565sub mon {
566 my ($noderef, $port) = split /#/, shift, 2;
567
568 my $node = $NODE{$noderef} || add_node $noderef;
569
570 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
571
572 unless (ref $cb) {
573 if (@_) {
574 # send a kill info message
575 my (@msg) = ($cb, @_);
576 $cb = sub { snd @msg, @_ };
577 } else {
578 # simply kill other port
579 my $port = $cb;
580 $cb = sub { kil $port, @_ if @_ };
581 }
582 }
583
584 $node->monitor ($port, $cb);
585
586 defined wantarray
587 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
233} 588}
234 589
235=item $port->destroy 590=item $guard = mon_guard $port, $ref, $ref...
236 591
237Explicitly destroy/remove/nuke/vaporise the port. 592Monitors the given C<$port> and keeps the passed references. When the port
593is killed, the references will be freed.
238 594
239Ports are normally kept alive by there mere existance alone, and need to 595Optionally returns a guard that will stop the monitoring.
240be destroyed explicitly.
241 596
242=cut 597This function is useful when you create e.g. timers or other watchers and
598want to free them when the port gets killed:
243 599
244sub destroy { 600 $port->rcv (start => sub {
245 my ($self) = @_; 601 my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
602 undef $timer if 0.9 < rand;
603 });
604 });
246 605
247 delete $AnyEvent::MP::Base::WKP{ $self->{wkname} }; 606=cut
248 607
249 delete $AnyEvent::MP::Base::PORT{$_} 608sub mon_guard {
250 for @{ $self->{names} }; 609 my ($port, @refs) = @_;
610
611 #TODO: mon-less form?
612
613 mon $port, sub { 0 && @refs }
251} 614}
252 615
253=back 616=item kil $port[, @reason]
254 617
255=head1 FUNCTIONS FOR NODES 618Kill the specified port with the given C<@reason>.
256 619
257=over 4 620If no C<@reason> is specified, then the port is killed "normally" (linked
621ports will not be kileld, or even notified).
258 622
259=item mon $noderef, $callback->($noderef, $status, $) 623Otherwise, linked ports get killed with the same reason (second form of
624C<mon>, see below).
260 625
261Monitors the given noderef. 626Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
627will be reported as reason C<< die => $@ >>.
262 628
263=item become_public endpoint... 629Transport/communication errors are reported as C<< transport_error =>
630$message >>.
264 631
265Tells the node to become a public node, i.e. reachable from other nodes.
266
267If no arguments are given, or the first argument is C<undef>, then
268AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
269local nodename resolves to.
270
271Otherwise the first argument must be an array-reference with transport
272endpoints ("ip:port", "hostname:port") or port numbers (in which case the
273local nodename is used as hostname). The endpoints are all resolved and
274will become the node reference.
275
276=cut 632=cut
633
634=item $port = spawn $node, $initfunc[, @initdata]
635
636Creates a port on the node C<$node> (which can also be a port ID, in which
637case it's the node where that port resides).
638
639The port ID of the newly created port is return immediately, and it is
640permissible to immediately start sending messages or monitor the port.
641
642After the port has been created, the init function is
643called. This function must be a fully-qualified function name
644(e.g. C<MyApp::Chat::Server::init>). To specify a function in the main
645program, use C<::name>.
646
647If the function doesn't exist, then the node tries to C<require>
648the package, then the package above the package and so on (e.g.
649C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
650exists or it runs out of package names.
651
652The init function is then called with the newly-created port as context
653object (C<$SELF>) and the C<@initdata> values as arguments.
654
655A common idiom is to pass your own port, monitor the spawned port, and
656in the init function, monitor the original port. This two-way monitoring
657ensures that both ports get cleaned up when there is a problem.
658
659Example: spawn a chat server port on C<$othernode>.
660
661 # this node, executed from within a port context:
662 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
663 mon $server;
664
665 # init function on C<$othernode>
666 sub connect {
667 my ($srcport) = @_;
668
669 mon $srcport;
670
671 rcv $SELF, sub {
672 ...
673 };
674 }
675
676=cut
677
678sub _spawn {
679 my $port = shift;
680 my $init = shift;
681
682 local $SELF = "$NODE#$port";
683 eval {
684 &{ load_func $init }
685 };
686 _self_die if $@;
687}
688
689sub spawn(@) {
690 my ($noderef, undef) = split /#/, shift, 2;
691
692 my $id = "$RUNIQ." . $ID++;
693
694 $_[0] =~ /::/
695 or Carp::croak "spawn init function must be a fully-qualified name, caught";
696
697 ($NODE{$noderef} || add_node $noderef)
698 ->send (["", "AnyEvent::MP::_spawn" => $id, @_]);
699
700 "$noderef#$id"
701}
277 702
278=back 703=back
279 704
280=head1 NODE MESSAGES 705=head1 NODE MESSAGES
281 706
282Nodes understand the following messages sent to them. Many of them take 707Nodes understand the following messages sent to them. Many of them take
283arguments called C<@reply>, which will simply be used to compose a reply 708arguments called C<@reply>, which will simply be used to compose a reply
284message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and 709message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
285the remaining arguments are simply the message data. 710the remaining arguments are simply the message data.
286 711
712While other messages exist, they are not public and subject to change.
713
287=over 4 714=over 4
288 715
289=cut 716=cut
290 717
291=item wkp => $name, @reply 718=item lookup => $name, @reply
292 719
293Replies with the port ID of the specified well-known port, or C<undef>. 720Replies with the port ID of the specified well-known port, or C<undef>.
294 721
295=item devnull => ... 722=item devnull => ...
296 723
319 snd $NODE, time => $myport, timereply => 1, 2; 746 snd $NODE, time => $myport, timereply => 1, 2;
320 # => snd $myport, timereply => 1, 2, <time> 747 # => snd $myport, timereply => 1, 2, <time>
321 748
322=back 749=back
323 750
751=head1 AnyEvent::MP vs. Distributed Erlang
752
753AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
754== aemp node, Erlang process == aemp port), so many of the documents and
755programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
756sample:
757
758 http://www.Erlang.se/doc/programming_rules.shtml
759 http://Erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
760 http://Erlang.org/download/Erlang-book-part1.pdf # chapters 5 and 6
761 http://Erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
762
763Despite the similarities, there are also some important differences:
764
765=over 4
766
767=item * Node references contain the recipe on how to contact them.
768
769Erlang relies on special naming and DNS to work everywhere in the
770same way. AEMP relies on each node knowing it's own address(es), with
771convenience functionality.
772
773This means that AEMP requires a less tightly controlled environment at the
774cost of longer node references and a slightly higher management overhead.
775
776=item * Erlang uses processes and a mailbox, AEMP does not queue.
777
778Erlang uses processes that selctively receive messages, and therefore
779needs a queue. AEMP is event based, queuing messages would serve no useful
780purpose.
781
782(But see L<Coro::MP> for a more Erlang-like process model on top of AEMP).
783
784=item * Erlang sends are synchronous, AEMP sends are asynchronous.
785
786Sending messages in Erlang is synchronous and blocks the process. AEMP
787sends are immediate, connection establishment is handled in the
788background.
789
790=item * Erlang can silently lose messages, AEMP cannot.
791
792Erlang makes few guarantees on messages delivery - messages can get lost
793without 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).
795
796AEMP guarantees correct ordering, and the guarantee that there are no
797holes in the message sequence.
798
799=item * In Erlang, processes can be declared dead and later be found to be
800alive.
801
802In Erlang it can happen that a monitored process is declared dead and
803linked processes get killed, but later it turns out that the process is
804still alive - and can receive messages.
805
806In AEMP, when port monitoring detects a port as dead, then that port will
807eventually be killed - it cannot happen that a node detects a port as dead
808and then later sends messages to it, finding it is still alive.
809
810=item * Erlang can send messages to the wrong port, AEMP does not.
811
812In Erlang it is quite possible that a node that restarts reuses a process
813ID known to other nodes for a completely different process, causing
814messages destined for that process to end up in an unrelated process.
815
816AEMP never reuses port IDs, so old messages or old port IDs floating
817around in the network will not be sent to an unrelated port.
818
819=item * Erlang uses unprotected connections, AEMP uses secure
820authentication and can use TLS.
821
822AEMP can use a proven protocol - SSL/TLS - to protect connections and
823securely authenticate nodes.
824
825=item * The AEMP protocol is optimised for both text-based and binary
826communications.
827
828The AEMP protocol, unlike the Erlang protocol, supports both
829language-independent text-only protocols (good for debugging) and binary,
830language-specific serialisers (e.g. Storable).
831
832It has also been carefully designed to be implementable in other languages
833with a minimum of work while gracefully degrading fucntionality to make the
834protocol simple.
835
836=item * AEMP has more flexible monitoring options than Erlang.
837
838In Erlang, you can chose to receive I<all> exit signals as messages
839or I<none>, there is no in-between, so monitoring single processes is
840difficult to implement. Monitoring in AEMP is more flexible than in
841Erlang, as one can choose between automatic kill, exit message or callback
842on a per-process basis.
843
844=item * Erlang tries to hide remote/local connections, AEMP does not.
845
846Monitoring in Erlang is not an indicator of process death/crashes,
847as linking is (except linking is unreliable in Erlang).
848
849In AEMP, you don't "look up" registered port names or send to named ports
850that might or might not be persistent. Instead, you normally spawn a port
851on the remote node. The init function monitors the you, and you monitor
852the remote port. Since both monitors are local to the node, they are much
853more reliable.
854
855This also saves round-trips and avoids sending messages to the wrong port
856(hard to do in Erlang).
857
858=back
859
324=head1 SEE ALSO 860=head1 SEE ALSO
325 861
326L<AnyEvent>. 862L<AnyEvent>.
327 863
328=head1 AUTHOR 864=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines