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.3 by root, Sat Aug 1 07:11:45 2009 UTC vs.
Revision 1.23 by root, Tue Aug 4 18:46:16 2009 UTC

4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::MP; 7 use AnyEvent::MP;
8 8
9 NODE # returns this node identifier
10 $NODE # contains this node identifier 9 $NODE # contains this node's noderef
10 NODE # returns this node's noderef
11 NODE $port # returns the noderef of the port
11 12
12 snd $port, type => data...; 13 snd $port, type => data...;
14
15 $SELF # receiving/own port id in rcv callbacks
13 16
14 rcv $port, smartmatch => $cb->($port, @msg); 17 rcv $port, smartmatch => $cb->($port, @msg);
15 18
16 # examples: 19 # examples:
17 rcv $port2, ping => sub { snd $_[0], "pong"; 0 }; 20 rcv $port2, ping => sub { snd $_[0], "pong"; 0 };
27This module (-family) implements a simple message passing framework. 30This module (-family) implements a simple message passing framework.
28 31
29Despite its simplicity, you can securely message other processes running 32Despite its simplicity, you can securely message other processes running
30on the same or other hosts. 33on the same or other hosts.
31 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
32=head1 CONCEPTS 42=head1 CONCEPTS
33 43
34=over 4 44=over 4
35 45
36=item port 46=item port
52 62
53Initially, nodes are either private (single-process only) or hidden 63Initially, nodes are either private (single-process only) or hidden
54(connected to a master node only). Only when they epxlicitly "become 64(connected to a master node only). Only when they epxlicitly "become
55public" can you send them messages from unrelated other nodes. 65public" can you send them messages from unrelated other nodes.
56 66
57=item noderef - C<host:port,host:port...>, C<id@noderef, C<id> 67=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
58 68
59A noderef is a string that either uniquely identifies a given node (for 69A noderef is a string that either uniquely identifies a given node (for
60private and hidden nodes), or contains a recipe on how to reach a given 70private and hidden nodes), or contains a recipe on how to reach a given
61node (for public nodes). 71node (for public nodes).
62 72
68 78
69=cut 79=cut
70 80
71package AnyEvent::MP; 81package AnyEvent::MP;
72 82
73use AnyEvent::MP::Util ();
74use AnyEvent::MP::Node; 83use AnyEvent::MP::Base;
75use AnyEvent::MP::Transport;
76 84
77use utf8;
78use common::sense; 85use common::sense;
79 86
80use Carp (); 87use Carp ();
81 88
82use AE (); 89use AE ();
83 90
84use base "Exporter"; 91use base "Exporter";
85 92
86our $VERSION = '0.0'; 93our $VERSION = '0.02';
87our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_); 94our @EXPORT = qw(
95 NODE $NODE *SELF node_of _any_
96 become_slave become_public
97 snd rcv mon kil reg psub
98 port
99);
88 100
89our $DEFAULT_SECRET; 101our $SELF;
90our $DEFAULT_PORT = "4040";
91 102
92our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 103sub _self_die() {
93our $CONNECT_TIMEOUT = 30; # includes handshake 104 my $msg = $@;
94 105 $msg =~ s/\n+$// unless ref $msg;
95sub default_secret { 106 kil $SELF, die => $msg;
96 unless (defined $DEFAULT_SECRET) {
97 if (open my $fh, "<$ENV{HOME}/.aemp-secret") {
98 sysread $fh, $DEFAULT_SECRET, -s $fh;
99 } else {
100 $DEFAULT_SECRET = AnyEvent::MP::Util::nonce 32;
101 }
102 }
103
104 $DEFAULT_SECRET
105} 107}
106 108
107=item NODE / $NODE 109=item $thisnode = NODE / $NODE
108 110
109The C<NODE ()> function and the C<$NODE> variable contain the noderef of 111The C<NODE> function returns, and the C<$NODE> variable contains
110the local node. The value is initialised by a call to C<become_public> or 112the noderef of the local node. The value is initialised by a call
111C<become_slave>, after which all local port identifiers become invalid. 113to C<become_public> or C<become_slave>, after which all local port
114identifiers become invalid.
112 115
113=cut 116=item $noderef = node_of $portid
114 117
115our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie 118Extracts and returns the noderef from a portid or a noderef.
116our $PUBLIC = 0;
117our $NODE;
118our $PORT;
119 119
120our %NODE; # node id to transport mapping, or "undef", for local node 120=item $SELF
121our %PORT; # local ports
122our %LISTENER; # local transports
123 121
124sub NODE() { $NODE } 122Contains the current port id while executing C<rcv> callbacks or C<psub>
123blocks.
125 124
126{ 125=item SELF, %SELF, @SELF...
127 use POSIX ();
128 my $nodename = (POSIX::uname)[1];
129 $NODE = "$$\@$nodename";
130}
131 126
132sub _ANY_() { 1 } 127Due to some quirks in how perl exports variables, it is impossible to
133sub _any_() { \&_ANY_ } 128just export C<$SELF>, all the symbols called C<SELF> are exported by this
134 129module, but only C<$SELF> is currently used.
135sub add_node {
136 my ($noderef) = @_;
137
138 return $NODE{$noderef}
139 if exists $NODE{$noderef};
140
141 for (split /,/, $noderef) {
142 return $NODE{$noderef} = $NODE{$_}
143 if exists $NODE{$_};
144 }
145
146 # for indirect sends, use a different class
147 my $node = new AnyEvent::MP::Node::Direct $noderef;
148
149 $NODE{$_} = $node
150 for $noderef, split /,/, $noderef;
151
152 $node
153}
154 130
155=item snd $portid, type => @data 131=item snd $portid, type => @data
156 132
157=item snd $portid, @msg 133=item snd $portid, @msg
158 134
159Send the given message to the given port ID, which can identify either a 135Send the given message to the given port ID, which can identify either
160local or a remote port. 136a local or a remote port, and can be either a string or soemthignt hat
137stringifies a sa port ID (such as a port object :).
161 138
162While the message can be about anything, it is highly recommended to use 139While the message can be about anything, it is highly recommended to use a
163a constant string as first element. 140string as first element (a portid, or some word that indicates a request
141type etc.).
164 142
165The message data effectively becomes read-only after a call to this 143The message data effectively becomes read-only after a call to this
166function: modifying any argument is not allowed and can cause many 144function: modifying any argument is not allowed and can cause many
167problems. 145problems.
168 146
170JSON is used, then only strings, numbers and arrays and hashes consisting 148JSON is used, then only strings, numbers and arrays and hashes consisting
171of those are allowed (no objects). When Storable is used, then anything 149of those are allowed (no objects). When Storable is used, then anything
172that Storable can serialise and deserialise is allowed, and for the local 150that Storable can serialise and deserialise is allowed, and for the local
173node, anything can be passed. 151node, anything can be passed.
174 152
175=cut 153=item kil $portid[, @reason]
176 154
177sub snd($@) { 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 {
178 my ($noderef, $port) = split /#/, shift, 2; 203 my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);
179 204
180 add_node $noderef 205 my $node = $NODE{$noderef} || add_node $noderef;
181 unless exists $NODE{$noderef};
182 206
183 $NODE{$noderef}->send (["$port", [@_]]); 207 #TODO: ports must not be references
184} 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 }
185 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
251
252Create a new local port object that supports message matching.
253
254=item $portid = port { my @msg = @_; $finished }
255
256Creates a "mini port", that is, a very lightweight port without any
257pattern matching behind it, and returns its ID.
258
259The 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
261will be destroyed. Otherwise it will stay alive.
262
263The message will be passed as-is, no extra argument (i.e. no port id) will
264be passed to the callback.
265
266If you need the local port id in the callback, this works nicely:
267
268 my $port; $port = miniport {
269 snd $otherport, reply => $port;
270 };
271
272=cut
273
274sub port(;&) {
275 my $id = "$UNIQ." . $ID++;
276 my $port = "$NODE#$id";
277
278 if (@_) {
279 my $cb = shift;
280 $PORT{$id} = sub {
281 local $SELF = $port;
282 eval {
283 &$cb
284 and kil $id;
285 };
286 _self_die if $@;
287 };
288 } else {
289 my $self = bless {
290 id => "$NODE#$id",
291 }, "AnyEvent::MP::Port";
292
293 $PORT_DATA{$id} = $self;
294 $PORT{$id} = sub {
295 local $SELF = $port;
296
297 eval {
298 for (@{ $self->{rc0}{$_[0]} }) {
299 $_ && &{$_->[0]}
300 && undef $_;
301 }
302
303 for (@{ $self->{rcv}{$_[0]} }) {
304 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
305 && &{$_->[0]}
306 && undef $_;
307 }
308
309 for (@{ $self->{any} }) {
310 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
311 && &{$_->[0]}
312 && undef $_;
313 }
314 };
315 _self_die if $@;
316 };
317 }
318
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
186=item rcv $portid, type => $callback->(@msg) 339=item rcv $portid, tagstring => $callback->(@msg), ...
187 340
188=item rcv $portid, $smartmatch => $callback->(@msg) 341=item rcv $portid, $smartmatch => $callback->(@msg), ...
189 342
190=item rcv $portid, [$smartmatch...] => $callback->(@msg) 343=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
191 344
192Register a callback on the port identified by C<$portid>, which I<must> be 345Register callbacks to be called on matching messages on the given port.
193a local port.
194 346
195The callback has to return a true value when its work is done, after 347The callback has to return a true value when its work is done, after
196which is will be removed, or a false value in which case it will stay 348which is will be removed, or a false value in which case it will stay
197registered. 349registered.
198 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
199If the match is an array reference, then it will be matched against the 357If the match is an array reference, then it will be matched against the
200first elements of the message, otherwise only the first element is being 358first elements of the message, otherwise only the first element is being
201matched. 359matched.
202 360
203Any element in the match that is specified as C<_any_> (a function 361Any element in the match that is specified as C<_any_> (a function
208also the most efficient match (by far). 366also the most efficient match (by far).
209 367
210=cut 368=cut
211 369
212sub rcv($@) { 370sub rcv($@) {
213 my ($port, $match, $cb) = @_;
214
215 my $port = $PORT{$port}
216 or do {
217 my ($noderef, $lport) = split /#/, $port; 371 my ($noderef, $port) = split /#/, shift, 2;
218 "AnyEvent::MP::Node::Self" eq ref $NODE{$noderef} 372
373 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
219 or Carp::croak "$port: can only rcv on local ports"; 374 or Carp::croak "$noderef#$port: rcv can only be called on local ports, caught";
220 375
221 $PORT{$lport} 376 my $self = $PORT_DATA{$port}
222 or Carp::croak "$port: port does not exist"; 377 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
223
224 $PORT{$port} = $PORT{$lport} # also return
225 };
226 378
379 "AnyEvent::MP::Port" eq ref $self
380 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
381
382 while (@_) {
383 my ($match, $cb) = splice @_, 0, 2;
384
227 if (!ref $match) { 385 if (!ref $match) {
228 push @{ $port->{rc0}{$match} }, [$cb]; 386 push @{ $self->{rc0}{$match} }, [$cb];
229 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 387 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
230 my ($type, @match) = @$match; 388 my ($type, @match) = @$match;
231 @match 389 @match
232 ? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match] 390 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
233 : push @{ $port->{rc0}{$match->[0]} }, [$cb]; 391 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
234 } else {
235 push @{ $port->{any} }, [$cb, $match];
236 }
237}
238
239sub _inject {
240 my ($port, $msg) = @{+shift};
241
242 $port = $PORT{$port}
243 or return;
244
245 @_ = @$msg;
246
247 for (@{ $port->{rc0}{$msg->[0]} }) {
248 $_ && &{$_->[0]}
249 && undef $_;
250 }
251
252 for (@{ $port->{rcv}{$msg->[0]} }) {
253 $_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1]
254 && &{$_->[0]}
255 && undef $_;
256 }
257
258 for (@{ $port->{any} }) {
259 $_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1]
260 && &{$_->[0]}
261 && undef $_;
262 }
263}
264
265sub normalise_noderef($) {
266 my ($noderef) = @_;
267
268 my $cv = AE::cv;
269 my @res;
270
271 $cv->begin (sub {
272 my %seen;
273 my @refs;
274 for (sort { $a->[0] <=> $b->[0] } @res) {
275 push @refs, $_->[1] unless $seen{$_->[1]}++
276 }
277 shift->send (join ",", @refs);
278 });
279
280 $noderef = $DEFAULT_PORT unless length $noderef;
281
282 my $idx;
283 for my $t (split /,/, $noderef) {
284 my $pri = ++$idx;
285
286 #TODO: this should be outside normalise_noderef and in become_public
287 if ($t =~ /^\d*$/) {
288 my $nodename = (POSIX::uname)[1];
289
290 $cv->begin;
291 AnyEvent::Socket::resolve_sockaddr $nodename, $t || "aemp=$DEFAULT_PORT", "tcp", 0, undef, sub {
292 for (@_) {
293 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
294 push @res, [
295 $pri += 1e-5,
296 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
297 ];
298 }
299 $cv->end;
300 };
301
302# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename;
303#
304# for (@ipv4) {
305# push @res, [
306# $pri,
307# AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $_, $t || $DEFAULT_PORT,
308# ];
309# }
310 } else { 392 } else {
311 my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT" 393 push @{ $self->{any} }, [$cb, $match];
312 or Carp::croak "$t: unparsable transport descriptor";
313
314 $cv->begin;
315 AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {
316 for (@_) {
317 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
318 push @res, [
319 $pri += 1e-5,
320 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
321 ];
322 }
323 $cv->end;
324 }
325 } 394 }
326 } 395 }
327
328 $cv->end;
329
330 $cv
331} 396}
332 397
333sub become_public { 398=item $closure = psub { BLOCK }
334 return if $PUBLIC;
335 399
336 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 400Remembers C<$SELF> and creates a closure out of the BLOCK. When the
337 my @args = @_; 401closure is executed, sets up the environment in the same way as in C<rcv>
402callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
338 403
339 $NODE = (normalise_noderef $noderef)->recv; 404This is useful when you register callbacks from C<rcv> callbacks:
340 405
341 for my $t (split /,/, $NODE) { 406 rcv delayed_reply => sub {
342 $NODE{$t} = $NODE{""}; 407 my ($delay, @reply) = @_;
343 408 my $timer = AE::timer $delay, 0, psub {
344 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 409 snd @reply, $SELF;
345
346 $LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port,
347 @args,
348 on_error => sub {
349 die "on_error<@_>\n";#d#
350 },
351 on_connect => sub {
352 my ($tp) = @_;
353
354 $NODE{$tp->{remote_id}} = $_[0];
355 },
356 sub {
357 my ($tp) = @_;
358
359 $NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp;
360 },
361 ; 410 };
411 };
412
413=cut
414
415sub psub(&) {
416 my $cb = shift;
417
418 my $port = $SELF
419 or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
420
421 sub {
422 local $SELF = $port;
423
424 if (wantarray) {
425 my @res = eval { &$cb };
426 _self_die if $@;
427 @res
428 } else {
429 my $res = eval { &$cb };
430 _self_die if $@;
431 $res
432 }
362 } 433 }
363
364 $PUBLIC = 1;
365} 434}
366 435
367############################################################################# 436=back
368# self node code
369 437
370sub _new_port($) { 438=head1 FUNCTIONS FOR NODES
371 my ($name) = @_;
372 439
373 my ($noderef, $portname) = split /#/, $name; 440=over 4
374 441
375 $PORT{$name} = 442=item become_public endpoint...
376 $PORT{$portname} = {
377 names => [$name, $portname],
378 };
379}
380 443
381$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE; 444Tells the node to become a public node, i.e. reachable from other nodes.
382_new_port "";
383 445
384rcv "", relay => \&snd; 446If no arguments are given, or the first argument is C<undef>, then
447AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
448local nodename resolves to.
449
450Otherwise the first argument must be an array-reference with transport
451endpoints ("ip:port", "hostname:port") or port numbers (in which case the
452local nodename is used as hostname). The endpoints are all resolved and
453will become the node reference.
454
455=cut
456
457=back
458
459=head1 NODE MESSAGES
460
461Nodes understand the following messages sent to them. Many of them take
462arguments 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
464the remaining arguments are simply the message data.
465
466=over 4
467
468=cut
469
470=item lookup => $name, @reply
471
472Replies with the port ID of the specified well-known port, or C<undef>.
473
474=item devnull => ...
475
476Generic data sink/CPU heat conversion.
477
478=item relay => $port, @msg
479
480Simply forwards the message to the given port.
481
482=item eval => $string[ @reply]
483
484Evaluates the given string. If C<@reply> is given, then a message of the
485form C<@reply, $@, @evalres> is sent.
486
487Example: crash another node.
488
489 snd $othernode, eval => "exit";
490
491=item time => @reply
492
493Replies the the current node time to C<@reply>.
494
495Example: tell the current node to send the current time to C<$myport> in a
496C<timereply> message.
497
498 snd $NODE, time => $myport, timereply => 1, 2;
499 # => snd $myport, timereply => 1, 2, <time>
385 500
386=back 501=back
387 502
388=head1 SEE ALSO 503=head1 SEE ALSO
389 504

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines