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

Diff Legend

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