ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/cvsroot/AnyEvent-MP/MP.pm
(Generate patch)

Comparing cvsroot/AnyEvent-MP/MP.pm (file contents):
Revision 1.2 by root, Fri Jul 31 20:55:46 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
37 41
38A port is something you can send messages to with the C<snd> function, and 42A port is something you can send messages to with the C<snd> function, and
39you can register C<rcv> handlers with. All C<rcv> handlers will receive 43you can register C<rcv> handlers with. All C<rcv> handlers will receive
40messages they match, messages will not be queued. 44messages they match, messages will not be queued.
41 45
42=item port id - C<pid@host#portname> 46=item port id - C<noderef#portname>
43 47
44A port id is always the node id, a hash-mark (C<#>) as separator, followed 48A port id is always the noderef, a hash-mark (C<#>) as separator, followed
45by a port name. 49by a port name (a printable string of unspecified format).
46
47A port name can be a well known port (basically an identifier/bareword),
48or a generated name, consisting of node id, a dot (C<.>), and an
49identifier.
50 50
51=item node 51=item node
52 52
53A node is a single process containing at least one port - the node 53A node is a single process containing at least one port - the node
54port. You can send messages to node ports to let them create new ports, 54port. You can send messages to node ports to let them create new ports,
55among other things. 55among other things.
56 56
57Initially, nodes are either private (single-process only) or hidden 57Initially, nodes are either private (single-process only) or hidden
58(connected to a father node only). Only when they epxlicitly "go public" 58(connected to a master node only). Only when they epxlicitly "become
59can you send them messages form unrelated other nodes. 59public" can you send them messages from unrelated other nodes.
60 60
61Public nodes automatically connect to all other public nodes in a network 61=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
62when they connect, creating a full mesh.
63 62
64=item node id - C<host:port>, C<id@host>, C<id>
65
66A node ID is a string that either uniquely identifies a given node (For 63A noderef is a string that either uniquely identifies a given node (for
67private and hidden nodes), or contains a recipe on how to reach a given 64private and hidden nodes), or contains a recipe on how to reach a given
68node (for public nodes). 65node (for public nodes).
69 66
70=back 67=back
71 68
72=head1 FUNCTIONS 69=head1 VARIABLES/FUNCTIONS
73 70
74=over 4 71=over 4
75 72
76=cut 73=cut
77 74
78package AnyEvent::MP; 75package AnyEvent::MP;
79 76
80use AnyEvent::MP::Util ();
81use AnyEvent::MP::Node; 77use AnyEvent::MP::Base;
82use AnyEvent::MP::Transport;
83 78
84use utf8;
85use common::sense; 79use common::sense;
86 80
87use Carp (); 81use Carp ();
88 82
89use AE (); 83use AE ();
90 84
91use base "Exporter"; 85use base "Exporter";
92 86
93our $VERSION = '0.0'; 87our $VERSION = '0.02';
94our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_); 88our @EXPORT = qw(
89 NODE $NODE $PORT snd rcv _any_
90 create_port create_port_on
91 create_miniport
92 become_slave become_public
93);
95 94
96our $DEFAULT_SECRET; 95=item NODE / $NODE
97our $DEFAULT_PORT = "4040";
98 96
99our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 97The C<NODE ()> function and the C<$NODE> variable contain the noderef of
100our $CONNECT_TIMEOUT = 30; # includes handshake 98the local node. The value is initialised by a call to C<become_public> or
99C<become_slave>, after which all local port identifiers become invalid.
101 100
102sub default_secret { 101=item snd $portid, type => @data
103 unless (defined $DEFAULT_SECRET) { 102
104 if (open my $fh, "<$ENV{HOME}/.aemp-secret") { 103=item snd $portid, @msg
105 sysread $fh, $DEFAULT_SECRET, -s $fh; 104
106 } else { 105Send the given message to the given port ID, which can identify either
107 $DEFAULT_SECRET = AnyEvent::MP::Util::nonce 32; 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 :).
108
109While the message can be about anything, it is highly recommended to use a
110string as first element (a portid, or some word that indicates a request
111type etc.).
112
113The message data effectively becomes read-only after a call to this
114function: modifying any argument is not allowed and can cause many
115problems.
116
117The type of data you can transfer depends on the transport protocol: when
118JSON is used, then only strings, numbers and arrays and hashes consisting
119of those are allowed (no objects). When Storable is used, then anything
120that Storable can serialise and deserialise is allowed, and for the local
121node, anything can be passed.
122
123=item $local_port = create_port
124
125Create a new local port object. See the next section for allowed methods.
126
127=cut
128
129sub create_port {
130 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
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 $_;
108 } 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
198=item $port->rcv (type => $callback->($port, @msg))
199
200=item $port->rcv ($smartmatch => $callback->($port, @msg))
201
202=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
203
204Register a callback on the given port.
205
206The callback has to return a true value when its work is done, after
207which is will be removed, or a false value in which case it will stay
208registered.
209
210If the match is an array reference, then it will be matched against the
211first elements of the message, otherwise only the first element is being
212matched.
213
214Any element in the match that is specified as C<_any_> (a function
215exported by this module) matches any single element of the message.
216
217While not required, it is highly recommended that the first matching
218element is a string identifying the message. The one-string-only match is
219also the most efficient match (by far).
220
221=cut
222
223sub rcv($@) {
224 my ($self, $match, $cb) = @_;
225
226 if (!ref $match) {
227 push @{ $self->{rc0}{$match} }, [$cb];
228 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
229 my ($type, @match) = @$match;
230 @match
231 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
232 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
233 } else {
234 push @{ $self->{any} }, [$cb, $match];
109 } 235 }
110
111 $DEFAULT_SECRET
112} 236}
113 237
114our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie 238=item $port->register ($name)
115our $PUBLIC = 0;
116our $NODE;
117our $PORT;
118 239
119our %NODE; # node id to transport mapping, or "undef", for local node 240Registers the given port under the well known name C<$name>. If the name
120our %PORT; # local ports 241already exists it is replaced.
121our %LISTENER; # local transports
122 242
123sub NODE() { $NODE } 243A port can only be registered under one well known name.
124 244
125{ 245=cut
126 use POSIX ();
127 my $nodename = (POSIX::uname)[1];
128 $NODE = "$$\@$nodename";
129}
130 246
131sub _ANY_() { 1 } 247sub register {
132sub _any_() { \&_ANY_ } 248 my ($self, $name) = @_;
133 249
134sub add_node { 250 $self->{wkname} = $name;
251 $AnyEvent::MP::Base::WKP{$name} = "$self";
252}
253
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 {
135 my ($noderef) = @_; 264 my ($self) = @_;
136 265
137 return $NODE{$noderef} 266 delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
138 if exists $NODE{$noderef};
139 267
140 for (split /,/, $noderef) { 268 delete $AnyEvent::MP::Base::PORT{$_}
141 return $NODE{$noderef} = $NODE{$_} 269 for @{ $self->{names} };
142 if exists $NODE{$_};
143 }
144
145 # for indirect sends, use a different class
146 my $node = new AnyEvent::MP::Node::Direct $noderef;
147
148 $NODE{$_} = $node
149 for $noderef, split /,/, $noderef;
150
151 $node
152} 270}
153 271
154sub snd($@) { 272=back
155 my ($noderef, $port) = split /#/, shift, 2;
156 273
157 add_node $noderef 274=head1 FUNCTIONS FOR NODES
158 unless exists $NODE{$noderef};
159 275
160 $NODE{$noderef}->send ([$port, [@_]]); 276=over 4
161}
162 277
163sub _inject { 278=item mon $noderef, $callback->($noderef, $status, $)
164 my ($port, $msg) = @{+shift};
165 279
166 $port = $PORT{$port} 280Monitors the given noderef.
167 or return;
168 281
169 use Data::Dumper; 282=item become_public endpoint...
170 warn Dumper $msg;
171}
172 283
173sub normalise_noderef($) { 284Tells the node to become a public node, i.e. reachable from other nodes.
174 my ($noderef) = @_;
175 285
176 my $cv = AE::cv; 286If no arguments are given, or the first argument is C<undef>, then
177 my @res; 287AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
288local nodename resolves to.
178 289
179 $cv->begin (sub { 290Otherwise the first argument must be an array-reference with transport
180 my %seen; 291endpoints ("ip:port", "hostname:port") or port numbers (in which case the
181 my @refs; 292local nodename is used as hostname). The endpoints are all resolved and
182 for (sort { $a->[0] <=> $b->[0] } @res) { 293will become the node reference.
183 push @refs, $_->[1] unless $seen{$_->[1]}++
184 }
185 shift->send (join ",", @refs);
186 });
187 294
188 $noderef = $DEFAULT_PORT unless length $noderef; 295=cut
189 296
190 my $idx; 297=back
191 for my $t (split /,/, $noderef) {
192 my $pri = ++$idx;
193
194 #TODO: this should be outside normalise_noderef and in become_public
195 if ($t =~ /^\d*$/) {
196 my $nodename = (POSIX::uname)[1];
197 298
198 $cv->begin; 299=head1 NODE MESSAGES
199 AnyEvent::Socket::resolve_sockaddr $nodename, $t || "aemp=$DEFAULT_PORT", "tcp", 0, undef, sub {
200 for (@_) {
201 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
202 push @res, [
203 $pri += 1e-5,
204 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
205 ];
206 }
207 $cv->end;
208 };
209 300
210# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename; 301Nodes understand the following messages sent to them. Many of them take
211# 302arguments called C<@reply>, which will simply be used to compose a reply
212# for (@ipv4) { 303message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
213# push @res, [ 304the remaining arguments are simply the message data.
214# $pri,
215# AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $_, $t || $DEFAULT_PORT,
216# ];
217# }
218 } else {
219 my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT"
220 or Carp::croak "$t: unparsable transport descriptor";
221 305
222 $cv->begin; 306=over 4
223 AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {
224 for (@_) {
225 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
226 push @res, [
227 $pri += 1e-5,
228 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
229 ];
230 }
231 $cv->end;
232 }
233 }
234 }
235 307
236 $cv->end; 308=cut
237 309
238 $cv 310=item wkp => $name, @reply
239}
240 311
241sub become_public { 312Replies with the port ID of the specified well-known port, or C<undef>.
242 return if $PUBLIC;
243 313
244 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 314=item devnull => ...
245 my @args = @_;
246 315
247 $NODE = (normalise_noderef $noderef)->recv; 316Generic data sink/CPU heat conversion.
248 317
249 my $self = new AnyEvent::MP::Node::Self noderef => $NODE; 318=item relay => $port, @msg
250 319
251 $NODE{""} = $self; # empty string == local node 320Simply forwards the message to the given port.
252 321
253 for my $t (split /,/, $NODE) { 322=item eval => $string[ @reply]
254 $NODE{$t} = $self;
255 323
256 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 324Evaluates the given string. If C<@reply> is given, then a message of the
325form C<@reply, $@, @evalres> is sent.
257 326
258 $LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port, 327Example: crash another node.
259 @args,
260 on_error => sub {
261 die "on_error<@_>\n";#d#
262 },
263 on_connect => sub {
264 my ($tp) = @_;
265 328
266 $NODE{$tp->{remote_id}} = $_[0]; 329 snd $othernode, eval => "exit";
267 },
268 sub {
269 my ($tp) = @_;
270 330
271 $NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp; 331=item time => @reply
272 },
273 ;
274 }
275 332
276 $PUBLIC = 1; 333Replies the the current node time to C<@reply>.
277} 334
335Example: tell the current node to send the current time to C<$myport> in a
336C<timereply> message.
337
338 snd $NODE, time => $myport, timereply => 1, 2;
339 # => snd $myport, timereply => 1, 2, <time>
278 340
279=back 341=back
280 342
281=head1 SEE ALSO 343=head1 SEE ALSO
282 344

Diff Legend

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