ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/cvsroot/AnyEvent-MP/MP.pm
(Generate patch)

Comparing cvsroot/AnyEvent-MP/MP.pm (file contents):
Revision 1.23 by root, Tue Aug 4 18:46:16 2009 UTC vs.
Revision 1.37 by root, Fri Aug 7 16:47:23 2009 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines