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.5 by root, Sat Aug 1 07:44:02 2009 UTC vs.
Revision 1.11 by root, Sun Aug 2 18:08:38 2009 UTC

… …
27This module (-family) implements a simple message passing framework. 27This module (-family) implements a simple message passing framework.
28 28
29Despite its simplicity, you can securely message other processes running 29Despite its simplicity, you can securely message other processes running
30on the same or other hosts. 30on the same or other hosts.
31 31
32At the moment, this module family is severly brokena nd underdocumented,
33so do not use. This was uploaded mainly to resreve the CPAN namespace -
34stay tuned!
35
32=head1 CONCEPTS 36=head1 CONCEPTS
33 37
34=over 4 38=over 4
35 39
36=item port 40=item port
… …
68 72
69=cut 73=cut
70 74
71package AnyEvent::MP; 75package AnyEvent::MP;
72 76
73use AnyEvent::MP::Util ();
74use AnyEvent::MP::Node; 77use AnyEvent::MP::Base;
75use AnyEvent::MP::Transport;
76 78
77use utf8;
78use common::sense; 79use common::sense;
79 80
80use Carp (); 81use Carp ();
81 82
82use AE (); 83use AE ();
83 84
84use base "Exporter"; 85use base "Exporter";
85 86
86our $VERSION = '0.0'; 87our $VERSION = '0.02';
87our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_); 88our @EXPORT = qw(
88 89 NODE $NODE $PORT snd rcv _any_
89our $DEFAULT_SECRET; 90 create_port create_port_on
90our $DEFAULT_PORT = "4040"; 91 create_miniport
91 92 become_slave become_public
92our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 93);
93our $CONNECT_TIMEOUT = 30; # includes handshake
94
95sub default_secret {
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}
106 94
107=item NODE / $NODE 95=item NODE / $NODE
108 96
109The 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
110the 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
111C<become_slave>, after which all local port identifiers become invalid. 99C<become_slave>, after which all local port identifiers become invalid.
112 100
113=cut
114
115our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie
116our $PUBLIC = 0;
117our $NODE;
118our $PORT;
119
120our %NODE; # node id to transport mapping, or "undef", for local node
121our %PORT; # local ports
122our %LISTENER; # local transports
123
124sub NODE() { $NODE }
125
126{
127 use POSIX ();
128 my $nodename = (POSIX::uname)[1];
129 $NODE = "$$\@$nodename";
130}
131
132sub _ANY_() { 1 }
133sub _any_() { \&_ANY_ }
134
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
155=item snd $portid, type => @data 101=item snd $portid, type => @data
156 102
157=item snd $portid, @msg 103=item snd $portid, @msg
158 104
159Send 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
160local 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 :).
161 108
162While 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
163a constant string as first element. 110string as first element (a portid, or some word that indicates a request
111type etc.).
164 112
165The message data effectively becomes read-only after a call to this 113The message data effectively becomes read-only after a call to this
166function: modifying any argument is not allowed and can cause many 114function: modifying any argument is not allowed and can cause many
167problems. 115problems.
168 116
… …
170JSON is used, then only strings, numbers and arrays and hashes consisting 118JSON is used, then only strings, numbers and arrays and hashes consisting
171of those are allowed (no objects). When Storable is used, then anything 119of those are allowed (no objects). When Storable is used, then anything
172that Storable can serialise and deserialise is allowed, and for the local 120that Storable can serialise and deserialise is allowed, and for the local
173node, anything can be passed. 121node, anything can be passed.
174 122
175=cut 123=item $local_port = create_port
176 124
177sub snd(@) { 125Create a new local port object. See the next section for allowed methods.
178 my ($noderef, $port) = split /#/, shift, 2;
179 126
180 add_node $noderef 127=cut
181 unless exists $NODE{$noderef};
182 128
183 $NODE{$noderef}->send (["$port", [@_]]); 129sub create_port {
184} 130 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
185 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
186=item rcv $portid, type => $callback->(@msg) 198=item $port->rcv (type => $callback->($port, @msg))
187 199
188=item rcv $portid, $smartmatch => $callback->(@msg) 200=item $port->rcv ($smartmatch => $callback->($port, @msg))
189 201
190=item rcv $portid, [$smartmatch...] => $callback->(@msg) 202=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
191 203
192Register a callback on the port identified by C<$portid>, which I<must> be 204Register a callback on the given port.
193a local port.
194 205
195The 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
196which 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
197registered. 208registered.
198 209
… …
208also the most efficient match (by far). 219also the most efficient match (by far).
209 220
210=cut 221=cut
211 222
212sub rcv($@) { 223sub rcv($@) {
213 my ($port, $match, $cb) = @_; 224 my ($self, $match, $cb) = @_;
214
215 my $port = $PORT{$port}
216 or do {
217 my ($noderef, $lport) = split /#/, $port;
218 "AnyEvent::MP::Node::Self" eq ref $NODE{$noderef}
219 or Carp::croak "$port: can only rcv on local ports";
220
221 $PORT{$lport}
222 or Carp::croak "$port: port does not exist";
223
224 $PORT{$port} = $PORT{$lport} # also return
225 };
226 225
227 if (!ref $match) { 226 if (!ref $match) {
228 push @{ $port->{rc0}{$match} }, [$cb]; 227 push @{ $self->{rc0}{$match} }, [$cb];
229 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 228 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
230 my ($type, @match) = @$match; 229 my ($type, @match) = @$match;
231 @match 230 @match
232 ? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match] 231 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
233 : push @{ $port->{rc0}{$match->[0]} }, [$cb]; 232 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
234 } else { 233 } else {
235 push @{ $port->{any} }, [$cb, $match]; 234 push @{ $self->{any} }, [$cb, $match];
236 } 235 }
237} 236}
238 237
239sub _inject { 238=item $port->register ($name)
240 my ($port, $msg) = @{+shift};
241 239
242 $port = $PORT{$port} 240Registers the given port under the well known name C<$name>. If the name
243 or return; 241already exists it is replaced.
244 242
245 @_ = @$msg; 243A port can only be registered under one well known name.
246 244
247 for (@{ $port->{rc0}{$msg->[0]} }) { 245=cut
248 $_ && &{$_->[0]}
249 && undef $_;
250 }
251 246
252 for (@{ $port->{rcv}{$msg->[0]} }) { 247sub register {
253 $_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1] 248 my ($self, $name) = @_;
254 && &{$_->[0]}
255 && undef $_;
256 }
257 249
258 for (@{ $port->{any} }) { 250 $self->{wkname} = $name;
259 $_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1] 251 $AnyEvent::MP::Base::WKP{$name} = "$self";
260 && &{$_->[0]}
261 && undef $_;
262 }
263} 252}
264 253
265sub 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 {
266 my ($noderef) = @_; 264 my ($self) = @_;
267 265
268 my $cv = AE::cv; 266 delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
269 my @res;
270 267
271 $cv->begin (sub { 268 delete $AnyEvent::MP::Base::PORT{$_}
272 my %seen; 269 for @{ $self->{names} };
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 {
311 my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT"
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 }
326 }
327
328 $cv->end;
329
330 $cv
331} 270}
332 271
333sub become_public { 272=back
334 return if $PUBLIC;
335 273
336 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 274=head1 FUNCTIONS FOR NODES
337 my @args = @_;
338 275
339 $NODE = (normalise_noderef $noderef)->recv; 276=over 4
340 277
341 for my $t (split /,/, $NODE) { 278=item mon $noderef, $callback->($noderef, $status, $)
342 $NODE{$t} = $NODE{""};
343 279
344 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 280Monitors the given noderef.
345 281
346 $LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port, 282=item become_public endpoint...
347 @args,
348 on_error => sub {
349 die "on_error<@_>\n";#d#
350 },
351 on_connect => sub {
352 my ($tp) = @_;
353 283
354 $NODE{$tp->{remote_id}} = $_[0]; 284Tells the node to become a public node, i.e. reachable from other nodes.
355 },
356 sub {
357 my ($tp) = @_;
358 285
359 $NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp; 286If no arguments are given, or the first argument is C<undef>, then
360 }, 287AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
361 ; 288local nodename resolves to.
362 }
363 289
364 $PUBLIC = 1; 290Otherwise the first argument must be an array-reference with transport
365} 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
366 296
367=back 297=back
368 298
369=head1 NODE MESSAGES 299=head1 NODE MESSAGES
370 300
… …
375 305
376=over 4 306=over 4
377 307
378=cut 308=cut
379 309
380############################################################################# 310=item wkp => $name, @reply
381# self node code
382 311
383sub _new_port($) { 312Replies with the port ID of the specified well-known port, or C<undef>.
384 my ($name) = @_;
385 313
386 my ($noderef, $portname) = split /#/, $name; 314=item devnull => ...
387 315
388 $PORT{$name} = 316Generic data sink/CPU heat conversion.
389 $PORT{$portname} = {
390 names => [$name, $portname],
391 };
392}
393
394$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE;
395_new_port "";
396 317
397=item relay => $port, @msg 318=item relay => $port, @msg
398 319
399Simply forwards the message to the given port. 320Simply forwards the message to the given port.
400
401=cut
402
403rcv "", relay => \&snd;
404 321
405=item eval => $string[ @reply] 322=item eval => $string[ @reply]
406 323
407Evaluates 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
408form C<@reply, $@, @evalres> is sent. 325form C<@reply, $@, @evalres> is sent.
409 326
410Example: crash another node. 327Example: crash another node.
411 328
412 snd $othernode, eval => "exit"; 329 snd $othernode, eval => "exit";
413 330
414=cut
415
416rcv "", eval => sub {
417 my (undef, $string, @reply) = @_;
418 my @res = eval $string;
419 snd @reply, "$@", @res if @reply;
420};
421
422=item time => @reply 331=item time => @reply
423 332
424Replies the the current node time to C<@reply>. 333Replies the the current node time to C<@reply>.
425 334
426Example: 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
427C<timereply> message. 336C<timereply> message.
428 337
429 snd $NODE, time => $myport, timereply => 1, 2; 338 snd $NODE, time => $myport, timereply => 1, 2;
430 # => snd $myport, timereply => 1, 2, <time> 339 # => snd $myport, timereply => 1, 2, <time>
431 340
432=cut
433
434rcv "", time => sub { shift; snd @_, AE::time };
435
436=back 341=back
437 342
438=head1 SEE ALSO 343=head1 SEE ALSO
439 344
440L<AnyEvent>. 345L<AnyEvent>.

Diff Legend

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