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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines