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.1 by root, Thu Jul 30 08:38:50 2009 UTC vs.
Revision 1.34 by root, Wed Aug 5 23:50:46 2009 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines