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.11 by root, Sun Aug 2 18:08:38 2009 UTC

… …
72 72
73=cut 73=cut
74 74
75package AnyEvent::MP; 75package AnyEvent::MP;
76 76
77use AnyEvent::MP::Util ();
78use AnyEvent::MP::Node; 77use AnyEvent::MP::Base;
79use AnyEvent::MP::Transport;
80 78
81use utf8;
82use common::sense; 79use common::sense;
83 80
84use Carp (); 81use Carp ();
85 82
86use AE (); 83use AE ();
87 84
88use base "Exporter"; 85use base "Exporter";
89 86
90our $VERSION = '0.01'; 87our $VERSION = '0.02';
91our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_); 88our @EXPORT = qw(
92 89 NODE $NODE $PORT snd rcv _any_
93our $DEFAULT_SECRET; 90 create_port create_port_on
94our $DEFAULT_PORT = "4040"; 91 create_miniport
95 92 become_slave become_public
96our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 93);
97our $CONNECT_TIMEOUT = 30; # includes handshake
98
99sub default_secret {
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}
110 94
111=item NODE / $NODE 95=item NODE / $NODE
112 96
113The C<NODE ()> function and the C<$NODE> variable contain the noderef of 97The C<NODE ()> function and the C<$NODE> variable contain the noderef of
114the local node. The value is initialised by a call to C<become_public> or 98the local node. The value is initialised by a call to C<become_public> or
115C<become_slave>, after which all local port identifiers become invalid. 99C<become_slave>, after which all local port identifiers become invalid.
116 100
117=cut
118
119our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie
120our $ID = "a0";
121our $PUBLIC = 0;
122our $NODE;
123our $PORT;
124
125our %NODE; # node id to transport mapping, or "undef", for local node
126our %PORT; # local ports
127our %LISTENER; # local transports
128
129sub NODE() { $NODE }
130
131{
132 use POSIX ();
133 my $nodename = (POSIX::uname)[1];
134 $NODE = "$$\@$nodename";
135}
136
137sub _ANY_() { 1 }
138sub _any_() { \&_ANY_ }
139
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
160=item snd $portid, type => @data 101=item snd $portid, type => @data
161 102
162=item snd $portid, @msg 103=item snd $portid, @msg
163 104
164Send the given message to the given port ID, which can identify either a 105Send the given message to the given port ID, which can identify either
165local or a remote port. 106a local or a remote port, and can be either a string or soemthignt hat
107stringifies a sa port ID (such as a port object :).
166 108
167While the message can be about anything, it is highly recommended to use 109While the message can be about anything, it is highly recommended to use a
168a constant string as first element. 110string as first element (a portid, or some word that indicates a request
111type etc.).
169 112
170The message data effectively becomes read-only after a call to this 113The message data effectively becomes read-only after a call to this
171function: modifying any argument is not allowed and can cause many 114function: modifying any argument is not allowed and can cause many
172problems. 115problems.
173 116
… …
175JSON is used, then only strings, numbers and arrays and hashes consisting 118JSON is used, then only strings, numbers and arrays and hashes consisting
176of those are allowed (no objects). When Storable is used, then anything 119of those are allowed (no objects). When Storable is used, then anything
177that Storable can serialise and deserialise is allowed, and for the local 120that Storable can serialise and deserialise is allowed, and for the local
178node, anything can be passed. 121node, anything can be passed.
179 122
180=cut 123=item $local_port = create_port
181 124
182sub snd(@) { 125Create a new local port object. See the next section for allowed methods.
183 my ($noderef, $port) = split /#/, shift, 2;
184 126
185 add_node $noderef 127=cut
186 unless exists $NODE{$noderef};
187 128
188 $NODE{$noderef}->send (["$port", [@_]]); 129sub create_port {
189} 130 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
190 131
132 my $self = bless {
133 id => "$NODE#$id",
134 names => [$id],
135 }, "AnyEvent::MP::Port";
136
137 $AnyEvent::MP::Base::PORT{$id} = sub {
138 unshift @_, $self;
139
140 for (@{ $self->{rc0}{$_[1]} }) {
141 $_ && &{$_->[0]}
142 && undef $_;
143 }
144
145 for (@{ $self->{rcv}{$_[1]} }) {
146 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
147 && &{$_->[0]}
148 && undef $_;
149 }
150
151 for (@{ $self->{any} }) {
152 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
153 && &{$_->[0]}
154 && undef $_;
155 }
156 };
157
158 $self
159}
160
161=item $portid = create_miniport { }
162
163Creates a "mini port", that is, a port without much #TODO
164
165=cut
166
167sub create_miniport(&) {
168 my $cb = shift;
169 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
170
171 $AnyEvent::MP::Base::PORT{$id} = sub {
172 &$cb
173 and delete $AnyEvent::MP::Base::PORT{$id};
174 };
175
176 "$NODE#$id"
177}
178
179package AnyEvent::MP::Port;
180
181=back
182
183=head1 METHODS FOR PORT OBJECTS
184
185=over 4
186
187=item "$port"
188
189A port object stringifies to its port ID, so can be used directly for
190C<snd> operations.
191
192=cut
193
194use overload
195 '""' => sub { $_[0]{id} },
196 fallback => 1;
197
191=item rcv $portid, type => $callback->(@msg) 198=item $port->rcv (type => $callback->($port, @msg))
192 199
193=item rcv $portid, $smartmatch => $callback->(@msg) 200=item $port->rcv ($smartmatch => $callback->($port, @msg))
194 201
195=item rcv $portid, [$smartmatch...] => $callback->(@msg) 202=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
196 203
197Register a callback on the port identified by C<$portid>, which I<must> be 204Register a callback on the given port.
198a local port.
199 205
200The callback has to return a true value when its work is done, after 206The 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 207which is will be removed, or a false value in which case it will stay
202registered. 208registered.
203 209
… …
213also the most efficient match (by far). 219also the most efficient match (by far).
214 220
215=cut 221=cut
216 222
217sub rcv($@) { 223sub rcv($@) {
218 my ($port, $match, $cb) = @_; 224 my ($self, $match, $cb) = @_;
219
220 my $port = $PORT{$port}
221 or do {
222 my ($noderef, $lport) = split /#/, $port;
223 "AnyEvent::MP::Node::Self" eq ref $NODE{$noderef}
224 or Carp::croak "$port: can only rcv on local ports";
225
226 $PORT{$lport}
227 or Carp::croak "$port: port does not exist";
228
229 $PORT{$port} = $PORT{$lport} # also return
230 };
231 225
232 if (!ref $match) { 226 if (!ref $match) {
233 push @{ $port->{rc0}{$match} }, [$cb]; 227 push @{ $self->{rc0}{$match} }, [$cb];
234 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 228 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
235 my ($type, @match) = @$match; 229 my ($type, @match) = @$match;
236 @match 230 @match
237 ? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match] 231 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
238 : push @{ $port->{rc0}{$match->[0]} }, [$cb]; 232 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
239 } else { 233 } else {
240 push @{ $port->{any} }, [$cb, $match]; 234 push @{ $self->{any} }, [$cb, $match];
241 } 235 }
242} 236}
243 237
244sub _inject { 238=item $port->register ($name)
245 my ($port, $msg) = @{+shift};
246 239
247 $port = $PORT{$port} 240Registers the given port under the well known name C<$name>. If the name
248 or return; 241already exists it is replaced.
249 242
250 @_ = @$msg; 243A port can only be registered under one well known name.
251 244
252 for (@{ $port->{rc0}{$msg->[0]} }) { 245=cut
253 $_ && &{$_->[0]}
254 && undef $_;
255 }
256 246
257 for (@{ $port->{rcv}{$msg->[0]} }) { 247sub register {
258 $_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1] 248 my ($self, $name) = @_;
259 && &{$_->[0]}
260 && undef $_;
261 }
262 249
263 for (@{ $port->{any} }) { 250 $self->{wkname} = $name;
264 $_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1] 251 $AnyEvent::MP::Base::WKP{$name} = "$self";
265 && &{$_->[0]}
266 && undef $_;
267 }
268} 252}
269 253
270sub normalise_noderef($) { 254=item $port->destroy
255
256Explicitly destroy/remove/nuke/vaporise the port.
257
258Ports are normally kept alive by there mere existance alone, and need to
259be destroyed explicitly.
260
261=cut
262
263sub destroy {
271 my ($noderef) = @_; 264 my ($self) = @_;
272 265
273 my $cv = AE::cv; 266 delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
274 my @res;
275 267
276 $cv->begin (sub { 268 delete $AnyEvent::MP::Base::PORT{$_}
277 my %seen; 269 for @{ $self->{names} };
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 {
316 my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT"
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 }
331 }
332
333 $cv->end;
334
335 $cv
336} 270}
337 271
338sub become_public { 272=back
339 return if $PUBLIC;
340 273
341 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 274=head1 FUNCTIONS FOR NODES
342 my @args = @_;
343 275
344 $NODE = (normalise_noderef $noderef)->recv; 276=over 4
345 277
346 for my $t (split /,/, $NODE) { 278=item mon $noderef, $callback->($noderef, $status, $)
347 $NODE{$t} = $NODE{""};
348 279
349 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 280Monitors the given noderef.
350 281
351 $LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port, 282=item become_public endpoint...
352 @args,
353 on_error => sub {
354 die "on_error<@_>\n";#d#
355 },
356 on_connect => sub {
357 my ($tp) = @_;
358 283
359 $NODE{$tp->{remote_id}} = $_[0]; 284Tells the node to become a public node, i.e. reachable from other nodes.
360 },
361 sub {
362 my ($tp) = @_;
363 285
364 $NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp; 286If no arguments are given, or the first argument is C<undef>, then
365 }, 287AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
366 ; 288local nodename resolves to.
367 }
368 289
369 $PUBLIC = 1; 290Otherwise the first argument must be an array-reference with transport
370} 291endpoints ("ip:port", "hostname:port") or port numbers (in which case the
292local nodename is used as hostname). The endpoints are all resolved and
293will become the node reference.
294
295=cut
371 296
372=back 297=back
373 298
374=head1 NODE MESSAGES 299=head1 NODE MESSAGES
375 300
… …
380 305
381=over 4 306=over 4
382 307
383=cut 308=cut
384 309
385############################################################################# 310=item wkp => $name, @reply
386# self node code
387 311
388sub _new_port($) { 312Replies 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 313
402=item devnull => ... 314=item devnull => ...
403 315
404Generic data sink/CPU heat conversion. 316Generic data sink/CPU heat conversion.
405 317
406=cut
407
408rcv "", devnull => sub { () };
409
410=item relay => $port, @msg 318=item relay => $port, @msg
411 319
412Simply forwards the message to the given port. 320Simply forwards the message to the given port.
413
414=cut
415
416rcv "", relay => sub { \&snd; () };
417 321
418=item eval => $string[ @reply] 322=item eval => $string[ @reply]
419 323
420Evaluates the given string. If C<@reply> is given, then a message of the 324Evaluates the given string. If C<@reply> is given, then a message of the
421form C<@reply, $@, @evalres> is sent. 325form C<@reply, $@, @evalres> is sent.
422 326
423Example: crash another node. 327Example: crash another node.
424 328
425 snd $othernode, eval => "exit"; 329 snd $othernode, eval => "exit";
426 330
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 331=item time => @reply
437 332
438Replies the the current node time to C<@reply>. 333Replies the the current node time to C<@reply>.
439 334
440Example: tell the current node to send the current time to C<$myport> in a 335Example: tell the current node to send the current time to C<$myport> in a
441C<timereply> message. 336C<timereply> message.
442 337
443 snd $NODE, time => $myport, timereply => 1, 2; 338 snd $NODE, time => $myport, timereply => 1, 2;
444 # => snd $myport, timereply => 1, 2, <time> 339 # => snd $myport, timereply => 1, 2, <time>
445 340
446=cut
447
448rcv "", time => sub { shift; snd @_, AE::time; () };
449
450=back 341=back
451 342
452=head1 SEE ALSO 343=head1 SEE ALSO
453 344
454L<AnyEvent>. 345L<AnyEvent>.

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines