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.6 by root, Sat Aug 1 10:02:33 2009 UTC vs.
Revision 1.22 by root, Tue Aug 4 18:33:30 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 };
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
32At the moment, this module family is severly brokena nd underdocumented, 35At the moment, this module family is severly brokena nd underdocumented,
33so do not use. This was uploaded mainly to resreve the CPAN namespace - 36so do not use. This was uploaded mainly to reserve the CPAN namespace -
34stay tuned! 37stay tuned!
35 38
36=head1 CONCEPTS 39=head1 CONCEPTS
37 40
38=over 4 41=over 4
72 75
73=cut 76=cut
74 77
75package AnyEvent::MP; 78package AnyEvent::MP;
76 79
77use AnyEvent::MP::Util ();
78use AnyEvent::MP::Node; 80use AnyEvent::MP::Base;
79use AnyEvent::MP::Transport;
80 81
81use utf8;
82use common::sense; 82use common::sense;
83 83
84use Carp (); 84use Carp ();
85 85
86use AE (); 86use AE ();
87 87
88use base "Exporter"; 88use base "Exporter";
89 89
90our $VERSION = '0.01'; 90our $VERSION = '0.02';
91our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_); 91our @EXPORT = qw(
92 NODE $NODE *SELF node_of _any_
93 become_slave become_public
94 snd rcv mon kil reg psub
95 port
96);
92 97
93our $DEFAULT_SECRET; 98our $SELF;
94our $DEFAULT_PORT = "4040";
95 99
96our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 100sub _self_die() {
97our $CONNECT_TIMEOUT = 30; # includes handshake 101 my $msg = $@;
98 102 $msg =~ s/\n+$// unless ref $msg;
99sub default_secret { 103 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} 104}
110 105
111=item NODE / $NODE 106=item $thisnode = NODE / $NODE
112 107
113The C<NODE ()> function and the C<$NODE> variable contain the noderef of 108The 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 109the noderef of the local node. The value is initialised by a call
115C<become_slave>, after which all local port identifiers become invalid. 110to C<become_public> or C<become_slave>, after which all local port
111identifiers become invalid.
116 112
117=cut 113=item $noderef = node_of $portid
118 114
119our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie 115Extracts and returns the noderef from a portid or a noderef.
120our $ID = "a0";
121our $PUBLIC = 0;
122our $NODE;
123our $PORT;
124 116
125our %NODE; # node id to transport mapping, or "undef", for local node 117=item $SELF
126our %PORT; # local ports
127our %LISTENER; # local transports
128 118
129sub NODE() { $NODE } 119Contains the current port id while executing C<rcv> callbacks or C<psub>
120blocks.
130 121
131{ 122=item SELF, %SELF, @SELF...
132 use POSIX ();
133 my $nodename = (POSIX::uname)[1];
134 $NODE = "$$\@$nodename";
135}
136 123
137sub _ANY_() { 1 } 124Due to some quirks in how perl exports variables, it is impossible to
138sub _any_() { \&_ANY_ } 125just export C<$SELF>, all the symbols called C<SELF> are exported by this
139 126module, 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 127
160=item snd $portid, type => @data 128=item snd $portid, type => @data
161 129
162=item snd $portid, @msg 130=item snd $portid, @msg
163 131
164Send the given message to the given port ID, which can identify either a 132Send the given message to the given port ID, which can identify either
165local or a remote port. 133a local or a remote port, and can be either a string or soemthignt hat
134stringifies a sa port ID (such as a port object :).
166 135
167While the message can be about anything, it is highly recommended to use 136While the message can be about anything, it is highly recommended to use a
168a constant string as first element. 137string as first element (a portid, or some word that indicates a request
138type etc.).
169 139
170The message data effectively becomes read-only after a call to this 140The message data effectively becomes read-only after a call to this
171function: modifying any argument is not allowed and can cause many 141function: modifying any argument is not allowed and can cause many
172problems. 142problems.
173 143
175JSON is used, then only strings, numbers and arrays and hashes consisting 145JSON is used, then only strings, numbers and arrays and hashes consisting
176of those are allowed (no objects). When Storable is used, then anything 146of those are allowed (no objects). When Storable is used, then anything
177that Storable can serialise and deserialise is allowed, and for the local 147that Storable can serialise and deserialise is allowed, and for the local
178node, anything can be passed. 148node, anything can be passed.
179 149
180=cut 150=item kil $portid[, @reason]
181 151
182sub snd(@) { 152Kill the specified port with the given C<@reason>.
153
154If no C<@reason> is specified, then the port is killed "normally" (linked
155ports will not be kileld, or even notified).
156
157Otherwise, linked ports get killed with the same reason (second form of
158C<mon>, see below).
159
160Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
161will be reported as reason C<< die => $@ >>.
162
163Transport/communication errors are reported as C<< transport_error =>
164$message >>.
165
166=item $guard = mon $portid, $cb->(@reason)
167
168=item $guard = mon $portid, $otherport
169
170=item $guard = mon $portid, $otherport, @msg
171
172Monitor the given port and do something when the port is killed.
173
174In the first form, the callback is simply called with any number
175of C<@reason> elements (no @reason means that the port was deleted
176"normally"). Note also that I<< the callback B<must> never die >>, so use
177C<eval> if unsure.
178
179In the second form, the other port will be C<kil>'ed with C<@reason>, iff
180a @reason was specified, i.e. on "normal" kils nothing happens, while
181under all other conditions, the other port is killed with the same reason.
182
183In the last form, a message of the form C<@msg, @reason> will be C<snd>.
184
185Example: call a given callback when C<$port> is killed.
186
187 mon $port, sub { warn "port died because of <@_>\n" };
188
189Example: kill ourselves when C<$port> is killed abnormally.
190
191 mon $port, $self;
192
193Example: send us a restart message another C<$port> is killed.
194
195 mon $port, $self => "restart";
196
197=cut
198
199sub mon {
183 my ($noderef, $port) = split /#/, shift, 2; 200 my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);
184 201
185 add_node $noderef 202 my $node = $NODE{$noderef} || add_node $noderef;
186 unless exists $NODE{$noderef};
187 203
188 $NODE{$noderef}->send (["$port", [@_]]); 204 #TODO: ports must not be references
189} 205 if (!ref $cb or "AnyEvent::MP::Port" eq ref $cb) {
206 if (@_) {
207 # send a kill info message
208 my (@msg) = ($cb, @_);
209 $cb = sub { snd @msg, @_ };
210 } else {
211 # simply kill other port
212 my $port = $cb;
213 $cb = sub { kil $port, @_ if @_ };
214 }
215 }
190 216
217 $node->monitor ($port, $cb);
218
219 defined wantarray
220 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
221}
222
223=item $guard = mon_guard $port, $ref, $ref...
224
225Monitors the given C<$port> and keeps the passed references. When the port
226is killed, the references will be freed.
227
228Optionally returns a guard that will stop the monitoring.
229
230This function is useful when you create e.g. timers or other watchers and
231want to free them when the port gets killed:
232
233 $port->rcv (start => sub {
234 my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
235 undef $timer if 0.9 < rand;
236 });
237 });
238
239=cut
240
241sub mon_guard {
242 my ($port, @refs) = @_;
243
244 mon $port, sub { 0 && @refs }
245}
246
247=item $local_port = port
248
249Create a new local port object that supports message matching.
250
251=item $portid = port { my @msg = @_; $finished }
252
253Creates a "mini port", that is, a very lightweight port without any
254pattern matching behind it, and returns its ID.
255
256The block will be called for every message received on the port. When the
257callback returns a true value its job is considered "done" and the port
258will be destroyed. Otherwise it will stay alive.
259
260The message will be passed as-is, no extra argument (i.e. no port id) will
261be passed to the callback.
262
263If you need the local port id in the callback, this works nicely:
264
265 my $port; $port = miniport {
266 snd $otherport, reply => $port;
267 };
268
269=cut
270
271sub port(;&) {
272 my $id = "$UNIQ." . $ID++;
273 my $port = "$NODE#$id";
274
275 if (@_) {
276 my $cb = shift;
277 $PORT{$id} = sub {
278 local $SELF = $port;
279 eval {
280 &$cb
281 and kil $id;
282 };
283 _self_die if $@;
284 };
285 } else {
286 my $self = bless {
287 id => "$NODE#$id",
288 }, "AnyEvent::MP::Port";
289
290 $PORT_DATA{$id} = $self;
291 $PORT{$id} = sub {
292 local $SELF = $port;
293
294 eval {
295 for (@{ $self->{rc0}{$_[0]} }) {
296 $_ && &{$_->[0]}
297 && undef $_;
298 }
299
300 for (@{ $self->{rcv}{$_[0]} }) {
301 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
302 && &{$_->[0]}
303 && undef $_;
304 }
305
306 for (@{ $self->{any} }) {
307 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
308 && &{$_->[0]}
309 && undef $_;
310 }
311 };
312 _self_die if $@;
313 };
314 }
315
316 $port
317}
318
319=item reg $portid, $name
320
321Registers the given port under the name C<$name>. If the name already
322exists it is replaced.
323
324A port can only be registered under one well known name.
325
326A port automatically becomes unregistered when it is killed.
327
328=cut
329
330sub reg(@) {
331 my ($portid, $name) = @_;
332
333 $REG{$name} = $portid;
334}
335
191=item rcv $portid, type => $callback->(@msg) 336=item rcv $portid, tagstring => $callback->(@msg), ...
192 337
193=item rcv $portid, $smartmatch => $callback->(@msg) 338=item rcv $portid, $smartmatch => $callback->(@msg), ...
194 339
195=item rcv $portid, [$smartmatch...] => $callback->(@msg) 340=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
196 341
197Register a callback on the port identified by C<$portid>, which I<must> be 342Register callbacks to be called on matching messages on the given port.
198a local port.
199 343
200The callback has to return a true value when its work is done, after 344The 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 345which is will be removed, or a false value in which case it will stay
202registered. 346registered.
203 347
348The global C<$SELF> (exported by this module) contains C<$portid> while
349executing the callback.
350
351Runtime errors wdurign callback execution will result in the port being
352C<kil>ed.
353
204If the match is an array reference, then it will be matched against the 354If 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 355first elements of the message, otherwise only the first element is being
206matched. 356matched.
207 357
208Any element in the match that is specified as C<_any_> (a function 358Any element in the match that is specified as C<_any_> (a function
213also the most efficient match (by far). 363also the most efficient match (by far).
214 364
215=cut 365=cut
216 366
217sub rcv($@) { 367sub rcv($@) {
218 my ($port, $match, $cb) = @_;
219
220 my $port = $PORT{$port}
221 or do {
222 my ($noderef, $lport) = split /#/, $port; 368 my ($noderef, $port) = split /#/, shift, 2;
223 "AnyEvent::MP::Node::Self" eq ref $NODE{$noderef} 369
370 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
224 or Carp::croak "$port: can only rcv on local ports"; 371 or Carp::croak "$noderef#$port: rcv can only be called on local ports, caught";
225 372
226 $PORT{$lport} 373 my $self = $PORT_DATA{$port}
227 or Carp::croak "$port: port does not exist"; 374 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 375
376 "AnyEvent::MP::Port" eq ref $self
377 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
378
379 while (@_) {
380 my ($match, $cb) = splice @_, 0, 2;
381
232 if (!ref $match) { 382 if (!ref $match) {
233 push @{ $port->{rc0}{$match} }, [$cb]; 383 push @{ $self->{rc0}{$match} }, [$cb];
234 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 384 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
235 my ($type, @match) = @$match; 385 my ($type, @match) = @$match;
236 @match 386 @match
237 ? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match] 387 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
238 : push @{ $port->{rc0}{$match->[0]} }, [$cb]; 388 : 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 { 389 } else {
316 my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT" 390 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 } 391 }
331 } 392 }
332
333 $cv->end;
334
335 $cv
336} 393}
337 394
338sub become_public { 395=item $closure = psub { BLOCK }
339 return if $PUBLIC;
340 396
341 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 397Remembers C<$SELF> and creates a closure out of the BLOCK. When the
342 my @args = @_; 398closure is executed, sets up the environment in the same way as in C<rcv>
399callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
343 400
344 $NODE = (normalise_noderef $noderef)->recv; 401This is useful when you register callbacks from C<rcv> callbacks:
345 402
346 for my $t (split /,/, $NODE) { 403 rcv delayed_reply => sub {
347 $NODE{$t} = $NODE{""}; 404 my ($delay, @reply) = @_;
348 405 my $timer = AE::timer $delay, 0, psub {
349 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 406 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 ; 407 };
408 };
409
410=cut
411
412sub psub(&) {
413 my $cb = shift;
414
415 my $port = $SELF
416 or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
417
418 sub {
419 local $SELF = $port;
420
421 if (wantarray) {
422 my @res = eval { &$cb };
423 _self_die if $@;
424 @res
425 } else {
426 my $res = eval { &$cb };
427 _self_die if $@;
428 $res
429 }
367 } 430 }
368
369 $PUBLIC = 1;
370} 431}
432
433=back
434
435=head1 FUNCTIONS FOR NODES
436
437=over 4
438
439=item become_public endpoint...
440
441Tells the node to become a public node, i.e. reachable from other nodes.
442
443If no arguments are given, or the first argument is C<undef>, then
444AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
445local nodename resolves to.
446
447Otherwise the first argument must be an array-reference with transport
448endpoints ("ip:port", "hostname:port") or port numbers (in which case the
449local nodename is used as hostname). The endpoints are all resolved and
450will become the node reference.
451
452=cut
371 453
372=back 454=back
373 455
374=head1 NODE MESSAGES 456=head1 NODE MESSAGES
375 457
380 462
381=over 4 463=over 4
382 464
383=cut 465=cut
384 466
385############################################################################# 467=item lookup => $name, @reply
386# self node code
387 468
388sub _new_port($) { 469Replies with the port ID of the specified well-known port, or C<undef>.
389 my ($name) = @_;
390 470
391 my ($noderef, $portname) = split /#/, $name; 471=item devnull => ...
392 472
393 $PORT{$name} = 473Generic data sink/CPU heat conversion.
394 $PORT{$portname} = {
395 names => [$name, $portname],
396 };
397}
398
399$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE;
400_new_port "";
401 474
402=item relay => $port, @msg 475=item relay => $port, @msg
403 476
404Simply forwards the message to the given port. 477Simply forwards the message to the given port.
405
406=cut
407
408rcv "", relay => \&snd;
409 478
410=item eval => $string[ @reply] 479=item eval => $string[ @reply]
411 480
412Evaluates the given string. If C<@reply> is given, then a message of the 481Evaluates the given string. If C<@reply> is given, then a message of the
413form C<@reply, $@, @evalres> is sent. 482form C<@reply, $@, @evalres> is sent.
414 483
415Example: crash another node. 484Example: crash another node.
416 485
417 snd $othernode, eval => "exit"; 486 snd $othernode, eval => "exit";
418 487
419=cut
420
421rcv "", eval => sub {
422 my (undef, $string, @reply) = @_;
423 my @res = eval $string;
424 snd @reply, "$@", @res if @reply;
425};
426
427=item time => @reply 488=item time => @reply
428 489
429Replies the the current node time to C<@reply>. 490Replies the the current node time to C<@reply>.
430 491
431Example: tell the current node to send the current time to C<$myport> in a 492Example: tell the current node to send the current time to C<$myport> in a
432C<timereply> message. 493C<timereply> message.
433 494
434 snd $NODE, time => $myport, timereply => 1, 2; 495 snd $NODE, time => $myport, timereply => 1, 2;
435 # => snd $myport, timereply => 1, 2, <time> 496 # => snd $myport, timereply => 1, 2, <time>
436 497
437=cut
438
439rcv "", time => sub { shift; snd @_, AE::time };
440
441=back 498=back
442 499
443=head1 SEE ALSO 500=head1 SEE ALSO
444 501
445L<AnyEvent>. 502L<AnyEvent>.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines