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.4 by root, Sat Aug 1 07:36:30 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=back 436=back
368 437
438=head1 FUNCTIONS FOR NODES
439
440=over 4
441
442=item become_public endpoint...
443
444Tells the node to become a public node, i.e. reachable from other nodes.
445
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
369=head1 NODE MESSAGES 459=head1 NODE MESSAGES
370 460
371Nodes understand the following messages sent to them: 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.
372 465
373=over 4 466=over 4
374 467
375=cut 468=cut
376 469
377############################################################################# 470=item lookup => $name, @reply
378# self node code
379 471
380sub _new_port($) { 472Replies with the port ID of the specified well-known port, or C<undef>.
381 my ($name) = @_;
382 473
383 my ($noderef, $portname) = split /#/, $name; 474=item devnull => ...
384 475
385 $PORT{$name} = 476Generic data sink/CPU heat conversion.
386 $PORT{$portname} = {
387 names => [$name, $portname],
388 };
389}
390
391$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE;
392_new_port "";
393 477
394=item relay => $port, @msg 478=item relay => $port, @msg
395 479
396Simply forwards the message to the given port. 480Simply forwards the message to the given port.
397 481
398=cut
399
400rcv "", relay => \&snd;
401
402=item eval => $string[ @reply] 482=item eval => $string[ @reply]
403 483
404Evaluates the given string. If C<@reply> is given, then a message of the 484Evaluates the given string. If C<@reply> is given, then a message of the
405form C<@reply, $@, @evalres> is sent (C<$reply[0]> is the port to reply to). 485form C<@reply, $@, @evalres> is sent.
406 486
407=cut 487Example: crash another node.
408 488
409rcv "", eval => sub { 489 snd $othernode, eval => "exit";
410 my (undef, $string, @reply) = @_;
411 my @res = eval $string;
412 snd @reply, "$@", @res if @reply;
413};
414 490
415=item time => @reply 491=item time => @reply
416 492
417Replies the the current node time to C<@reply>. 493Replies the the current node time to C<@reply>.
418 494
419=cut 495Example: tell the current node to send the current time to C<$myport> in a
496C<timereply> message.
420 497
421rcv "", time => sub { shift; snd @_, AE::time }; 498 snd $NODE, time => $myport, timereply => 1, 2;
499 # => snd $myport, timereply => 1, 2, <time>
422 500
423=back 501=back
424 502
425=head1 SEE ALSO 503=head1 SEE ALSO
426 504

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines