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.10 by root, Sun Aug 2 18:05:43 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 become_slave become_public
92);
95 93
96our $DEFAULT_SECRET; 94=item NODE / $NODE
97our $DEFAULT_PORT = "4040";
98 95
99our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 96The C<NODE ()> function and the C<$NODE> variable contain the noderef of
100our $CONNECT_TIMEOUT = 30; # includes handshake 97the local node. The value is initialised by a call to C<become_public> or
98C<become_slave>, after which all local port identifiers become invalid.
101 99
102sub default_secret { 100=item snd $portid, type => @data
103 unless (defined $DEFAULT_SECRET) { 101
104 if (open my $fh, "<$ENV{HOME}/.aemp-secret") { 102=item snd $portid, @msg
105 sysread $fh, $DEFAULT_SECRET, -s $fh; 103
106 } else { 104Send the given message to the given port ID, which can identify either
107 $DEFAULT_SECRET = AnyEvent::MP::Util::nonce 32; 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 :).
107
108While the message can be about anything, it is highly recommended to use a
109string as first element (a portid, or some word that indicates a request
110type etc.).
111
112The message data effectively becomes read-only after a call to this
113function: modifying any argument is not allowed and can cause many
114problems.
115
116The type of data you can transfer depends on the transport protocol: when
117JSON is used, then only strings, numbers and arrays and hashes consisting
118of those are allowed (no objects). When Storable is used, then anything
119that Storable can serialise and deserialise is allowed, and for the local
120node, anything can be passed.
121
122=item $local_port = create_port
123
124Create a new local port object. See the next section for allowed methods.
125
126=cut
127
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 $_;
108 } 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
158}
159
160=item $portid = create_miniport { }
161
162Creates a "mini port", that is, a port without much #TODO
163
164=cut
165
166sub create_miniport(&) {
167 my $cb = shift;
168 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
169
170 $AnyEvent::MP::Base::PORT{$id} = sub {
171 &$cb
172 and delete $AnyEvent::MP::Base::PORT{$id};
173 };
174
175 "$NODE#$id"
176}
177
178package AnyEvent::MP::Port;
179
180=back
181
182=head1 METHODS FOR PORT OBJECTS
183
184=over 4
185
186=item "$port"
187
188A port object stringifies to its port ID, so can be used directly for
189C<snd> operations.
190
191=cut
192
193use overload
194 '""' => sub { $_[0]{id} },
195 fallback => 1;
196
197=item $port->rcv (type => $callback->($port, @msg))
198
199=item $port->rcv ($smartmatch => $callback->($port, @msg))
200
201=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
202
203Register a callback on the given port.
204
205The callback has to return a true value when its work is done, after
206which is will be removed, or a false value in which case it will stay
207registered.
208
209If the match is an array reference, then it will be matched against the
210first elements of the message, otherwise only the first element is being
211matched.
212
213Any element in the match that is specified as C<_any_> (a function
214exported by this module) matches any single element of the message.
215
216While not required, it is highly recommended that the first matching
217element is a string identifying the message. The one-string-only match is
218also the most efficient match (by far).
219
220=cut
221
222sub rcv($@) {
223 my ($self, $match, $cb) = @_;
224
225 if (!ref $match) {
226 push @{ $self->{rc0}{$match} }, [$cb];
227 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
228 my ($type, @match) = @$match;
229 @match
230 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
231 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
232 } else {
233 push @{ $self->{any} }, [$cb, $match];
109 } 234 }
110
111 $DEFAULT_SECRET
112} 235}
113 236
114our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie 237=item $port->register ($name)
115our $PUBLIC = 0;
116our $NODE;
117our $PORT;
118 238
119our %NODE; # node id to transport mapping, or "undef", for local node 239Registers the given port under the well known name C<$name>. If the name
120our %PORT; # local ports 240already exists it is replaced.
121our %LISTENER; # local transports
122 241
123sub NODE() { $NODE } 242A port can only be registered under one well known name.
124 243
125{ 244=cut
126 use POSIX ();
127 my $nodename = (POSIX::uname)[1];
128 $NODE = "$$\@$nodename";
129}
130 245
131sub _ANY_() { 1 } 246sub register {
132sub _any_() { \&_ANY_ } 247 my ($self, $name) = @_;
133 248
134sub add_node { 249 $self->{wkname} = $name;
250 $AnyEvent::MP::Base::WKP{$name} = "$self";
251}
252
253=item $port->destroy
254
255Explicitly destroy/remove/nuke/vaporise the port.
256
257Ports are normally kept alive by there mere existance alone, and need to
258be destroyed explicitly.
259
260=cut
261
262sub destroy {
135 my ($noderef) = @_; 263 my ($self) = @_;
136 264
137 return $NODE{$noderef} 265 delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
138 if exists $NODE{$noderef};
139 266
140 for (split /,/, $noderef) { 267 delete $AnyEvent::MP::Base::PORT{$_}
141 return $NODE{$noderef} = $NODE{$_} 268 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} 269}
153 270
154sub snd($@) { 271=back
155 my ($noderef, $port) = split /#/, shift, 2;
156 272
157 add_node $noderef 273=head1 FUNCTIONS FOR NODES
158 unless exists $NODE{$noderef};
159 274
160 $NODE{$noderef}->send ([$port, [@_]]); 275=over 4
161}
162 276
163sub _inject { 277=item mon $noderef, $callback->($noderef, $status, $)
164 my ($port, $msg) = @{+shift};
165 278
166 $port = $PORT{$port} 279Monitors the given noderef.
167 or return;
168 280
169 use Data::Dumper; 281=item become_public endpoint...
170 warn Dumper $msg;
171}
172 282
173sub normalise_noderef($) { 283Tells the node to become a public node, i.e. reachable from other nodes.
174 my ($noderef) = @_;
175 284
176 my $cv = AE::cv; 285If no arguments are given, or the first argument is C<undef>, then
177 my @res; 286AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
287local nodename resolves to.
178 288
179 $cv->begin (sub { 289Otherwise the first argument must be an array-reference with transport
180 my %seen; 290endpoints ("ip:port", "hostname:port") or port numbers (in which case the
181 my @refs; 291local nodename is used as hostname). The endpoints are all resolved and
182 for (sort { $a->[0] <=> $b->[0] } @res) { 292will become the node reference.
183 push @refs, $_->[1] unless $seen{$_->[1]}++
184 }
185 shift->send (join ",", @refs);
186 });
187 293
188 $noderef = $DEFAULT_PORT unless length $noderef; 294=cut
189 295
190 my $idx; 296=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 297
198 $cv->begin; 298=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 299
210# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename; 300Nodes understand the following messages sent to them. Many of them take
211# 301arguments called C<@reply>, which will simply be used to compose a reply
212# for (@ipv4) { 302message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
213# push @res, [ 303the 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 304
222 $cv->begin; 305=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 306
236 $cv->end; 307=cut
237 308
238 $cv 309=item wkp => $name, @reply
239}
240 310
241sub become_public { 311Replies with the port ID of the specified well-known port, or C<undef>.
242 return if $PUBLIC;
243 312
244 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 313=item devnull => ...
245 my @args = @_;
246 314
247 $NODE = (normalise_noderef $noderef)->recv; 315Generic data sink/CPU heat conversion.
248 316
249 my $self = new AnyEvent::MP::Node::Self noderef => $NODE; 317=item relay => $port, @msg
250 318
251 $NODE{""} = $self; # empty string == local node 319Simply forwards the message to the given port.
252 320
253 for my $t (split /,/, $NODE) { 321=item eval => $string[ @reply]
254 $NODE{$t} = $self;
255 322
256 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 323Evaluates the given string. If C<@reply> is given, then a message of the
324form C<@reply, $@, @evalres> is sent.
257 325
258 $LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port, 326Example: crash another node.
259 @args,
260 on_error => sub {
261 die "on_error<@_>\n";#d#
262 },
263 on_connect => sub {
264 my ($tp) = @_;
265 327
266 $NODE{$tp->{remote_id}} = $_[0]; 328 snd $othernode, eval => "exit";
267 },
268 sub {
269 my ($tp) = @_;
270 329
271 $NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp; 330=item time => @reply
272 },
273 ;
274 }
275 331
276 $PUBLIC = 1; 332Replies the the current node time to C<@reply>.
277} 333
334Example: tell the current node to send the current time to C<$myport> in a
335C<timereply> message.
336
337 snd $NODE, time => $myport, timereply => 1, 2;
338 # => snd $myport, timereply => 1, 2, <time>
278 339
279=back 340=back
280 341
281=head1 SEE ALSO 342=head1 SEE ALSO
282 343

Diff Legend

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