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.7 by root, Sat Aug 1 15:04: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
32At the moment, this module family is severly brokena nd underdocumented, 38At the moment, this module family is severly broken and underdocumented,
33so do not use. This was uploaded mainly to resreve the CPAN namespace - 39so do not use. This was uploaded mainly to reserve the CPAN namespace -
34stay tuned! 40stay tuned! The basic API should be finished, however.
35 41
36=head1 CONCEPTS 42=head1 CONCEPTS
37 43
38=over 4 44=over 4
39 45
72 78
73=cut 79=cut
74 80
75package AnyEvent::MP; 81package AnyEvent::MP;
76 82
77use AnyEvent::MP::Util ();
78use AnyEvent::MP::Node; 83use AnyEvent::MP::Base;
79use AnyEvent::MP::Transport;
80 84
81use utf8;
82use common::sense; 85use common::sense;
83 86
84use Carp (); 87use Carp ();
85 88
86use AE (); 89use AE ();
87 90
88use base "Exporter"; 91use base "Exporter";
89 92
90our $VERSION = '0.01'; 93our $VERSION = '0.02';
91our @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);
92 100
93our $DEFAULT_SECRET; 101our $SELF;
94our $DEFAULT_PORT = "4040";
95 102
96our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 103sub _self_die() {
97our $CONNECT_TIMEOUT = 30; # includes handshake 104 my $msg = $@;
98 105 $msg =~ s/\n+$// unless ref $msg;
99sub default_secret { 106 kil $SELF, die => $msg;
100 unless (defined $DEFAULT_SECRET) {
101 if (open my $fh, "<$ENV{HOME}/.aemp-secret") {
102 sysread $fh, $DEFAULT_SECRET, -s $fh;
103 } else {
104 $DEFAULT_SECRET = AnyEvent::MP::Util::nonce 32;
105 }
106 }
107
108 $DEFAULT_SECRET
109} 107}
110 108
111=item NODE / $NODE 109=item $thisnode = NODE / $NODE
112 110
113The C<NODE ()> function and the C<$NODE> variable contain the noderef of 111The C<NODE> function returns, and the C<$NODE> variable contains
114the 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
115C<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.
116 115
117=cut 116=item $noderef = node_of $portid
118 117
119our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie 118Extracts and returns the noderef from a portid or a noderef.
120our $ID = "a0";
121our $PUBLIC = 0;
122our $NODE;
123our $PORT;
124 119
125our %NODE; # node id to transport mapping, or "undef", for local node 120=item $SELF
126our %PORT; # local ports
127our %LISTENER; # local transports
128 121
129sub NODE() { $NODE } 122Contains the current port id while executing C<rcv> callbacks or C<psub>
123blocks.
130 124
131{ 125=item SELF, %SELF, @SELF...
132 use POSIX ();
133 my $nodename = (POSIX::uname)[1];
134 $NODE = "$$\@$nodename";
135}
136 126
137sub _ANY_() { 1 } 127Due to some quirks in how perl exports variables, it is impossible to
138sub _any_() { \&_ANY_ } 128just export C<$SELF>, all the symbols called C<SELF> are exported by this
139 129module, but only C<$SELF> is currently used.
140sub add_node {
141 my ($noderef) = @_;
142
143 return $NODE{$noderef}
144 if exists $NODE{$noderef};
145
146 for (split /,/, $noderef) {
147 return $NODE{$noderef} = $NODE{$_}
148 if exists $NODE{$_};
149 }
150
151 # for indirect sends, use a different class
152 my $node = new AnyEvent::MP::Node::Direct $noderef;
153
154 $NODE{$_} = $node
155 for $noderef, split /,/, $noderef;
156
157 $node
158}
159 130
160=item snd $portid, type => @data 131=item snd $portid, type => @data
161 132
162=item snd $portid, @msg 133=item snd $portid, @msg
163 134
164Send 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
165local 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 :).
166 138
167While 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
168a constant string as first element. 140string as first element (a portid, or some word that indicates a request
141type etc.).
169 142
170The message data effectively becomes read-only after a call to this 143The message data effectively becomes read-only after a call to this
171function: modifying any argument is not allowed and can cause many 144function: modifying any argument is not allowed and can cause many
172problems. 145problems.
173 146
175JSON is used, then only strings, numbers and arrays and hashes consisting 148JSON is used, then only strings, numbers and arrays and hashes consisting
176of those are allowed (no objects). When Storable is used, then anything 149of those are allowed (no objects). When Storable is used, then anything
177that Storable can serialise and deserialise is allowed, and for the local 150that Storable can serialise and deserialise is allowed, and for the local
178node, anything can be passed. 151node, anything can be passed.
179 152
180=cut 153=item kil $portid[, @reason]
181 154
182sub 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 {
183 my ($noderef, $port) = split /#/, shift, 2; 203 my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);
184 204
185 add_node $noderef 205 my $node = $NODE{$noderef} || add_node $noderef;
186 unless exists $NODE{$noderef};
187 206
188 $NODE{$noderef}->send (["$port", [@_]]); 207 #TODO: ports must not be references
189} 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 }
190 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
191=item rcv $portid, type => $callback->(@msg) 339=item rcv $portid, tagstring => $callback->(@msg), ...
192 340
193=item rcv $portid, $smartmatch => $callback->(@msg) 341=item rcv $portid, $smartmatch => $callback->(@msg), ...
194 342
195=item rcv $portid, [$smartmatch...] => $callback->(@msg) 343=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
196 344
197Register 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.
198a local port.
199 346
200The 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
201which 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
202registered. 349registered.
203 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
204If 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
205first elements of the message, otherwise only the first element is being 358first elements of the message, otherwise only the first element is being
206matched. 359matched.
207 360
208Any 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
213also the most efficient match (by far). 366also the most efficient match (by far).
214 367
215=cut 368=cut
216 369
217sub rcv($@) { 370sub rcv($@) {
218 my ($port, $match, $cb) = @_;
219
220 my $port = $PORT{$port}
221 or do {
222 my ($noderef, $lport) = split /#/, $port; 371 my ($noderef, $port) = split /#/, shift, 2;
223 "AnyEvent::MP::Node::Self" eq ref $NODE{$noderef} 372
373 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
224 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";
225 375
226 $PORT{$lport} 376 my $self = $PORT_DATA{$port}
227 or Carp::croak "$port: port does not exist"; 377 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
228
229 $PORT{$port} = $PORT{$lport} # also return
230 };
231 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
232 if (!ref $match) { 385 if (!ref $match) {
233 push @{ $port->{rc0}{$match} }, [$cb]; 386 push @{ $self->{rc0}{$match} }, [$cb];
234 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 387 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
235 my ($type, @match) = @$match; 388 my ($type, @match) = @$match;
236 @match 389 @match
237 ? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match] 390 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
238 : push @{ $port->{rc0}{$match->[0]} }, [$cb]; 391 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
239 } else {
240 push @{ $port->{any} }, [$cb, $match];
241 }
242}
243
244sub _inject {
245 my ($port, $msg) = @{+shift};
246
247 $port = $PORT{$port}
248 or return;
249
250 @_ = @$msg;
251
252 for (@{ $port->{rc0}{$msg->[0]} }) {
253 $_ && &{$_->[0]}
254 && undef $_;
255 }
256
257 for (@{ $port->{rcv}{$msg->[0]} }) {
258 $_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1]
259 && &{$_->[0]}
260 && undef $_;
261 }
262
263 for (@{ $port->{any} }) {
264 $_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1]
265 && &{$_->[0]}
266 && undef $_;
267 }
268}
269
270sub normalise_noderef($) {
271 my ($noderef) = @_;
272
273 my $cv = AE::cv;
274 my @res;
275
276 $cv->begin (sub {
277 my %seen;
278 my @refs;
279 for (sort { $a->[0] <=> $b->[0] } @res) {
280 push @refs, $_->[1] unless $seen{$_->[1]}++
281 }
282 shift->send (join ",", @refs);
283 });
284
285 $noderef = $DEFAULT_PORT unless length $noderef;
286
287 my $idx;
288 for my $t (split /,/, $noderef) {
289 my $pri = ++$idx;
290
291 #TODO: this should be outside normalise_noderef and in become_public
292 if ($t =~ /^\d*$/) {
293 my $nodename = (POSIX::uname)[1];
294
295 $cv->begin;
296 AnyEvent::Socket::resolve_sockaddr $nodename, $t || "aemp=$DEFAULT_PORT", "tcp", 0, undef, sub {
297 for (@_) {
298 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
299 push @res, [
300 $pri += 1e-5,
301 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
302 ];
303 }
304 $cv->end;
305 };
306
307# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename;
308#
309# for (@ipv4) {
310# push @res, [
311# $pri,
312# AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $_, $t || $DEFAULT_PORT,
313# ];
314# }
315 } else { 392 } else {
316 my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT" 393 push @{ $self->{any} }, [$cb, $match];
317 or Carp::croak "$t: unparsable transport descriptor";
318
319 $cv->begin;
320 AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {
321 for (@_) {
322 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
323 push @res, [
324 $pri += 1e-5,
325 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
326 ];
327 }
328 $cv->end;
329 }
330 } 394 }
331 } 395 }
332
333 $cv->end;
334
335 $cv
336} 396}
337 397
338sub become_public { 398=item $closure = psub { BLOCK }
339 return if $PUBLIC;
340 399
341 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 400Remembers C<$SELF> and creates a closure out of the BLOCK. When the
342 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.
343 403
344 $NODE = (normalise_noderef $noderef)->recv; 404This is useful when you register callbacks from C<rcv> callbacks:
345 405
346 for my $t (split /,/, $NODE) { 406 rcv delayed_reply => sub {
347 $NODE{$t} = $NODE{""}; 407 my ($delay, @reply) = @_;
348 408 my $timer = AE::timer $delay, 0, psub {
349 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 409 snd @reply, $SELF;
350
351 $LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port,
352 @args,
353 on_error => sub {
354 die "on_error<@_>\n";#d#
355 },
356 on_connect => sub {
357 my ($tp) = @_;
358
359 $NODE{$tp->{remote_id}} = $_[0];
360 },
361 sub {
362 my ($tp) = @_;
363
364 $NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp;
365 },
366 ; 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 }
367 } 433 }
368
369 $PUBLIC = 1;
370} 434}
435
436=back
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
371 456
372=back 457=back
373 458
374=head1 NODE MESSAGES 459=head1 NODE MESSAGES
375 460
380 465
381=over 4 466=over 4
382 467
383=cut 468=cut
384 469
385############################################################################# 470=item lookup => $name, @reply
386# self node code
387 471
388sub _new_port($) { 472Replies with the port ID of the specified well-known port, or C<undef>.
389 my ($name) = @_;
390
391 my ($noderef, $portname) = split /#/, $name;
392
393 $PORT{$name} =
394 $PORT{$portname} = {
395 names => [$name, $portname],
396 };
397}
398
399$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE;
400_new_port "";
401 473
402=item devnull => ... 474=item devnull => ...
403 475
404Generic data sink/CPU heat conversion. 476Generic data sink/CPU heat conversion.
405 477
406=cut
407
408rcv "", devnull => sub { () };
409
410=item relay => $port, @msg 478=item relay => $port, @msg
411 479
412Simply forwards the message to the given port. 480Simply forwards the message to the given port.
413
414=cut
415
416rcv "", relay => sub { \&snd; () };
417 481
418=item eval => $string[ @reply] 482=item eval => $string[ @reply]
419 483
420Evaluates 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
421form C<@reply, $@, @evalres> is sent. 485form C<@reply, $@, @evalres> is sent.
422 486
423Example: crash another node. 487Example: crash another node.
424 488
425 snd $othernode, eval => "exit"; 489 snd $othernode, eval => "exit";
426 490
427=cut
428
429rcv "", eval => sub {
430 my (undef, $string, @reply) = @_;
431 my @res = eval $string;
432 snd @reply, "$@", @res if @reply;
433 ()
434};
435
436=item time => @reply 491=item time => @reply
437 492
438Replies the the current node time to C<@reply>. 493Replies the the current node time to C<@reply>.
439 494
440Example: tell the current node to send the current time to C<$myport> in a 495Example: tell the current node to send the current time to C<$myport> in a
441C<timereply> message. 496C<timereply> message.
442 497
443 snd $NODE, time => $myport, timereply => 1, 2; 498 snd $NODE, time => $myport, timereply => 1, 2;
444 # => snd $myport, timereply => 1, 2, <time> 499 # => snd $myport, timereply => 1, 2, <time>
445 500
446=cut
447
448rcv "", time => sub { shift; snd @_, AE::time; () };
449
450=back 501=back
451 502
452=head1 SEE ALSO 503=head1 SEE ALSO
453 504
454L<AnyEvent>. 505L<AnyEvent>.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines