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.22 by root, Tue Aug 4 18:33:30 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
33on the same or other hosts. 38on the same or other hosts.
34 39
40For an introduction to this module family, see the L<AnyEvent::MP::Intro>
41manual page.
42
35At the moment, this module family is severly brokena nd underdocumented, 43At the moment, this module family is severly broken and underdocumented,
36so do not use. This was uploaded mainly to reserve the CPAN namespace - 44so do not use. This was uploaded mainly to reserve the CPAN namespace -
37stay tuned! 45stay tuned! The basic API should be finished, however.
38 46
39=head1 CONCEPTS 47=head1 CONCEPTS
40 48
41=over 4 49=over 4
42 50
43=item port 51=item port
44 52
45A 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).
46you can register C<rcv> handlers with. All C<rcv> handlers will receive 54
47messages 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.
48 58
49=item port id - C<noderef#portname> 59=item port id - C<noderef#portname>
50 60
51A 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
52by 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.
53 65
54=item node 66=item node
55 67
56A 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
57port. 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
58among other things. 70create new ports, among other things.
59 71
60Initially, nodes are either private (single-process only) or hidden 72Nodes are either private (single-process only), slaves (connected to a
61(connected to a master node only). Only when they epxlicitly "become 73master node only) or public nodes (connectable from unrelated nodes).
62public" can you send them messages from unrelated other nodes.
63 74
64=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>
65 76
66A 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
67private 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
68node (for public nodes). 79node (for public nodes).
69 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
70=back 90=back
71 91
72=head1 VARIABLES/FUNCTIONS 92=head1 VARIABLES/FUNCTIONS
73 93
74=over 4 94=over 4
85 105
86use AE (); 106use AE ();
87 107
88use base "Exporter"; 108use base "Exporter";
89 109
90our $VERSION = '0.02'; 110our $VERSION = '0.1';
91our @EXPORT = qw( 111our @EXPORT = qw(
92 NODE $NODE *SELF node_of _any_ 112 NODE $NODE *SELF node_of _any_
93 become_slave become_public 113 resolve_node initialise_node
94 snd rcv mon kil reg psub 114 snd rcv mon kil reg psub
95 port 115 port
96); 116);
97 117
98our $SELF; 118our $SELF;
108The C<NODE> function returns, and the C<$NODE> variable contains 128The C<NODE> function returns, and the C<$NODE> variable contains
109the 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
110to 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
111identifiers become invalid. 131identifiers become invalid.
112 132
113=item $noderef = node_of $portid 133=item $noderef = node_of $port
114 134
115Extracts 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
116 229
117=item $SELF 230=item $SELF
118 231
119Contains 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>
120blocks. 233blocks.
123 236
124Due 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
125just 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
126module, but only C<$SELF> is currently used. 239module, but only C<$SELF> is currently used.
127 240
128=item snd $portid, type => @data 241=item snd $port, type => @data
129 242
130=item snd $portid, @msg 243=item snd $port, @msg
131 244
132Send 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
133a 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
134stringifies a sa port ID (such as a port object :). 247stringifies a sa port ID (such as a port object :).
135 248
145JSON is used, then only strings, numbers and arrays and hashes consisting 258JSON is used, then only strings, numbers and arrays and hashes consisting
146of those are allowed (no objects). When Storable is used, then anything 259of those are allowed (no objects). When Storable is used, then anything
147that Storable can serialise and deserialise is allowed, and for the local 260that Storable can serialise and deserialise is allowed, and for the local
148node, anything can be passed. 261node, anything can be passed.
149 262
150=item kil $portid[, @reason]
151
152Kill the specified port with the given C<@reason>.
153
154If no C<@reason> is specified, then the port is killed "normally" (linked
155ports will not be kileld, or even notified).
156
157Otherwise, linked ports get killed with the same reason (second form of
158C<mon>, see below).
159
160Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
161will be reported as reason C<< die => $@ >>.
162
163Transport/communication errors are reported as C<< transport_error =>
164$message >>.
165
166=item $guard = mon $portid, $cb->(@reason)
167
168=item $guard = mon $portid, $otherport
169
170=item $guard = mon $portid, $otherport, @msg
171
172Monitor the given port and do something when the port is killed.
173
174In the first form, the callback is simply called with any number
175of C<@reason> elements (no @reason means that the port was deleted
176"normally"). Note also that I<< the callback B<must> never die >>, so use
177C<eval> if unsure.
178
179In the second form, the other port will be C<kil>'ed with C<@reason>, iff
180a @reason was specified, i.e. on "normal" kils nothing happens, while
181under all other conditions, the other port is killed with the same reason.
182
183In the last form, a message of the form C<@msg, @reason> will be C<snd>.
184
185Example: call a given callback when C<$port> is killed.
186
187 mon $port, sub { warn "port died because of <@_>\n" };
188
189Example: kill ourselves when C<$port> is killed abnormally.
190
191 mon $port, $self;
192
193Example: send us a restart message another C<$port> is killed.
194
195 mon $port, $self => "restart";
196
197=cut
198
199sub mon {
200 my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);
201
202 my $node = $NODE{$noderef} || add_node $noderef;
203
204 #TODO: ports must not be references
205 if (!ref $cb or "AnyEvent::MP::Port" eq ref $cb) {
206 if (@_) {
207 # send a kill info message
208 my (@msg) = ($cb, @_);
209 $cb = sub { snd @msg, @_ };
210 } else {
211 # simply kill other port
212 my $port = $cb;
213 $cb = sub { kil $port, @_ if @_ };
214 }
215 }
216
217 $node->monitor ($port, $cb);
218
219 defined wantarray
220 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
221}
222
223=item $guard = mon_guard $port, $ref, $ref...
224
225Monitors the given C<$port> and keeps the passed references. When the port
226is killed, the references will be freed.
227
228Optionally returns a guard that will stop the monitoring.
229
230This function is useful when you create e.g. timers or other watchers and
231want to free them when the port gets killed:
232
233 $port->rcv (start => sub {
234 my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
235 undef $timer if 0.9 < rand;
236 });
237 });
238
239=cut
240
241sub mon_guard {
242 my ($port, @refs) = @_;
243
244 mon $port, sub { 0 && @refs }
245}
246
247=item $local_port = port 263=item $local_port = port
248 264
249Create 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.
250 268
251=item $portid = port { my @msg = @_; $finished } 269=item $port = port { my @msg = @_; $finished }
252 270
253Creates a "mini port", that is, a very lightweight port without any 271Creates a "miniport", that is, a very lightweight port without any pattern
254pattern 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.
255 274
256The 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
257callback 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
258will be destroyed. Otherwise it will stay alive. 277will be destroyed. Otherwise it will stay alive.
259 278
260The 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
261be passed to the callback. 280be passed to the callback.
262 281
263If 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:
264 283
265 my $port; $port = miniport { 284 my $port; $port = port {
266 snd $otherport, reply => $port; 285 snd $otherport, reply => $port;
267 }; 286 };
268 287
269=cut 288=cut
289
290sub rcv($@);
270 291
271sub port(;&) { 292sub port(;&) {
272 my $id = "$UNIQ." . $ID++; 293 my $id = "$UNIQ." . $ID++;
273 my $port = "$NODE#$id"; 294 my $port = "$NODE#$id";
274 295
275 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) {
276 my $cb = shift; 385 my $cb = shift;
386 delete $PORT_DATA{$portid};
277 $PORT{$id} = sub { 387 $PORT{$portid} = sub {
278 local $SELF = $port; 388 local $SELF = $port;
279 eval { 389 eval {
280 &$cb 390 &$cb
281 and kil $id; 391 and kil $port;
282 }; 392 };
283 _self_die if $@; 393 _self_die if $@;
284 }; 394 };
285 } else { 395 } else {
396 my $self = $PORT_DATA{$portid} ||= do {
286 my $self = bless { 397 my $self = bless {
287 id => "$NODE#$id", 398 id => $port,
288 }, "AnyEvent::MP::Port"; 399 }, "AnyEvent::MP::Port";
289 400
290 $PORT_DATA{$id} = $self;
291 $PORT{$id} = sub { 401 $PORT{$portid} = sub {
292 local $SELF = $port; 402 local $SELF = $port;
293 403
294 eval { 404 eval {
295 for (@{ $self->{rc0}{$_[0]} }) { 405 for (@{ $self->{rc0}{$_[0]} }) {
296 $_ && &{$_->[0]} 406 $_ && &{$_->[0]}
297 && undef $_; 407 && undef $_;
298 } 408 }
299 409
300 for (@{ $self->{rcv}{$_[0]} }) { 410 for (@{ $self->{rcv}{$_[0]} }) {
301 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1] 411 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
302 && &{$_->[0]} 412 && &{$_->[0]}
303 && undef $_; 413 && undef $_;
304 } 414 }
305 415
306 for (@{ $self->{any} }) { 416 for (@{ $self->{any} }) {
307 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1] 417 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
308 && &{$_->[0]} 418 && &{$_->[0]}
309 && undef $_; 419 && undef $_;
420 }
310 } 421 };
422 _self_die if $@;
311 }; 423 };
312 _self_die if $@; 424
425 $self
313 }; 426 };
314 }
315 427
316 $port
317}
318
319=item reg $portid, $name
320
321Registers the given port under the name C<$name>. If the name already
322exists it is replaced.
323
324A port can only be registered under one well known name.
325
326A port automatically becomes unregistered when it is killed.
327
328=cut
329
330sub reg(@) {
331 my ($portid, $name) = @_;
332
333 $REG{$name} = $portid;
334}
335
336=item rcv $portid, tagstring => $callback->(@msg), ...
337
338=item rcv $portid, $smartmatch => $callback->(@msg), ...
339
340=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
341
342Register callbacks to be called on matching messages on the given port.
343
344The callback has to return a true value when its work is done, after
345which is will be removed, or a false value in which case it will stay
346registered.
347
348The global C<$SELF> (exported by this module) contains C<$portid> while
349executing the callback.
350
351Runtime errors wdurign callback execution will result in the port being
352C<kil>ed.
353
354If the match is an array reference, then it will be matched against the
355first elements of the message, otherwise only the first element is being
356matched.
357
358Any element in the match that is specified as C<_any_> (a function
359exported by this module) matches any single element of the message.
360
361While not required, it is highly recommended that the first matching
362element is a string identifying the message. The one-string-only match is
363also the most efficient match (by far).
364
365=cut
366
367sub rcv($@) {
368 my ($noderef, $port) = split /#/, shift, 2;
369
370 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
371 or Carp::croak "$noderef#$port: rcv can only be called on local ports, caught";
372
373 my $self = $PORT_DATA{$port}
374 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
375
376 "AnyEvent::MP::Port" eq ref $self 428 "AnyEvent::MP::Port" eq ref $self
377 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";
378 430
379 while (@_) { 431 while (@_) {
380 my ($match, $cb) = splice @_, 0, 2; 432 my ($match, $cb) = splice @_, 0, 2;
381 433
382 if (!ref $match) { 434 if (!ref $match) {
383 push @{ $self->{rc0}{$match} }, [$cb]; 435 push @{ $self->{rc0}{$match} }, [$cb];
384 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 436 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
385 my ($type, @match) = @$match; 437 my ($type, @match) = @$match;
386 @match 438 @match
387 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match] 439 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
388 : push @{ $self->{rc0}{$match->[0]} }, [$cb]; 440 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
389 } else { 441 } else {
390 push @{ $self->{any} }, [$cb, $match]; 442 push @{ $self->{any} }, [$cb, $match];
443 }
391 } 444 }
392 } 445 }
446
447 $port
393} 448}
394 449
395=item $closure = psub { BLOCK } 450=item $closure = psub { BLOCK }
396 451
397Remembers 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
428 $res 483 $res
429 } 484 }
430 } 485 }
431} 486}
432 487
433=back 488=item $guard = mon $port, $cb->(@reason)
434 489
435=head1 FUNCTIONS FOR NODES 490=item $guard = mon $port, $rcvport
436 491
437=over 4 492=item $guard = mon $port
438 493
439=item become_public endpoint... 494=item $guard = mon $port, $rcvport, @msg
440 495
441Tells 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.
442 498
443If no arguments are given, or the first argument is C<undef>, then 499In the first form (callback), the callback is simply called with any
444AnyEvent::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
445local nodename resolves to. 501"normally"). Note also that I<< the callback B<must> never die >>, so use
502C<eval> if unsure.
446 503
447Otherwise the first argument must be an array-reference with transport 504In the second form (another port given), the other port (C<$rcvport)
448endpoints ("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
449local nodename is used as hostname). The endpoints are all resolved and 506"normal" kils nothing happens, while under all other conditions, the other
450will become the node reference. 507port is killed with the same reason.
451 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
452=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 >>.
453 602
454=back 603=back
455 604
456=head1 NODE MESSAGES 605=head1 NODE MESSAGES
457 606
458Nodes understand the following messages sent to them. Many of them take 607Nodes understand the following messages sent to them. Many of them take
459arguments 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
460message - 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
461the remaining arguments are simply the message data. 610the remaining arguments are simply the message data.
462 611
612While other messages exist, they are not public and subject to change.
613
463=over 4 614=over 4
464 615
465=cut 616=cut
466 617
467=item lookup => $name, @reply 618=item lookup => $name, @reply
495 snd $NODE, time => $myport, timereply => 1, 2; 646 snd $NODE, time => $myport, timereply => 1, 2;
496 # => snd $myport, timereply => 1, 2, <time> 647 # => snd $myport, timereply => 1, 2, <time>
497 648
498=back 649=back
499 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
500=head1 SEE ALSO 760=head1 SEE ALSO
501 761
502L<AnyEvent>. 762L<AnyEvent>.
503 763
504=head1 AUTHOR 764=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines