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

Diff Legend

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