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

Diff Legend

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