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.7 by root, Sat Aug 1 15:04:30 2009 UTC vs.
Revision 1.122 by root, Wed Feb 29 18:44:59 2012 UTC

1=head1 NAME 1=head1 NAME
2 2
3AnyEvent::MP - multi-processing/message-passing framework 3AnyEvent::MP - erlang-style multi-processing/message-passing framework
4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::MP; 7 use AnyEvent::MP;
8 8
9 NODE # returns this node identifier
10 $NODE # contains this node identifier 9 $NODE # contains this node's node ID
10 NODE # returns this node's node ID
11 11
12 $SELF # receiving/own port id in rcv callbacks
13
14 # initialise the node so it can send/receive messages
15 configure;
16
17 # ports are message destinations
18
19 # sending messages
12 snd $port, type => data...; 20 snd $port, type => data...;
21 snd $port, @msg;
22 snd @msg_with_first_element_being_a_port;
13 23
14 rcv $port, smartmatch => $cb->($port, @msg); 24 # creating/using ports, the simple way
25 my $simple_port = port { my @msg = @_ };
15 26
16 # examples: 27 # creating/using ports, tagged message matching
28 my $port = port;
17 rcv $port2, ping => sub { snd $_[0], "pong"; 0 }; 29 rcv $port, ping => sub { snd $_[0], "pong" };
18 rcv $port1, pong => sub { warn "pong received\n" }; 30 rcv $port, pong => sub { warn "pong received\n" };
19 snd $port2, ping => $port1;
20 31
21 # more, smarter, matches (_any_ is exported by this module) 32 # create a port on another node
22 rcv $port, [child_died => $pid] => sub { ... 33 my $port = spawn $node, $initfunc, @initdata;
23 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3 34
35 # destroy a port again
36 kil $port; # "normal" kill
37 kil $port, my_error => "everything is broken"; # error kill
38
39 # monitoring
40 mon $localport, $cb->(@msg) # callback is invoked on death
41 mon $localport, $otherport # kill otherport on abnormal death
42 mon $localport, $otherport, @msg # send message on death
43
44 # temporarily execute code in port context
45 peval $port, sub { die "kill the port!" };
46
47 # execute callbacks in $SELF port context
48 my $timer = AE::timer 1, 0, psub {
49 die "kill the port, delayed";
50 };
51
52=head1 CURRENT STATUS
53
54 bin/aemp - stable.
55 AnyEvent::MP - stable API, should work.
56 AnyEvent::MP::Intro - explains most concepts.
57 AnyEvent::MP::Kernel - mostly stable API.
58 AnyEvent::MP::Global - stable API.
24 59
25=head1 DESCRIPTION 60=head1 DESCRIPTION
26 61
27This module (-family) implements a simple message passing framework. 62This module (-family) implements a simple message passing framework.
28 63
29Despite its simplicity, you can securely message other processes running 64Despite its simplicity, you can securely message other processes running
30on the same or other hosts. 65on the same or other hosts, and you can supervise entities remotely.
31 66
32At the moment, this module family is severly brokena nd underdocumented, 67For an introduction to this module family, see the L<AnyEvent::MP::Intro>
33so do not use. This was uploaded mainly to resreve the CPAN namespace - 68manual page and the examples under F<eg/>.
34stay tuned!
35 69
36=head1 CONCEPTS 70=head1 CONCEPTS
37 71
38=over 4 72=over 4
39 73
40=item port 74=item port
41 75
42A port is something you can send messages to with the C<snd> function, and 76Not to be confused with a TCP port, a "port" is something you can send
43you can register C<rcv> handlers with. All C<rcv> handlers will receive 77messages to (with the C<snd> function).
44messages they match, messages will not be queued.
45 78
79Ports allow you to register C<rcv> handlers that can match all or just
80some messages. Messages send to ports will not be queued, regardless of
81anything was listening for them or not.
82
83Ports are represented by (printable) strings called "port IDs".
84
46=item port id - C<noderef#portname> 85=item port ID - C<nodeid#portname>
47 86
48A port id is always the noderef, a hash-mark (C<#>) as separator, followed 87A port ID is the concatenation of a node ID, a hash-mark (C<#>) as
49by a port name (a printable string of unspecified format). 88separator, and a port name (a printable string of unspecified format).
50 89
51=item node 90=item node
52 91
53A node is a single process containing at least one port - the node 92A node is a single process containing at least one port - the node port,
54port. You can send messages to node ports to let them create new ports, 93which enables nodes to manage each other remotely, and to create new
55among other things. 94ports.
56 95
57Initially, nodes are either private (single-process only) or hidden 96Nodes are either public (have one or more listening ports) or private
58(connected to a master node only). Only when they epxlicitly "become 97(no listening ports). Private nodes cannot talk to other private nodes
59public" can you send them messages from unrelated other nodes. 98currently, but all nodes can talk to public nodes.
60 99
61=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id> 100Nodes is represented by (printable) strings called "node IDs".
62 101
102=item node ID - C<[A-Za-z0-9_\-.:]*>
103
63A noderef is a string that either uniquely identifies a given node (for 104A node ID is a string that uniquely identifies the node within a
64private and hidden nodes), or contains a recipe on how to reach a given 105network. Depending on the configuration used, node IDs can look like a
65node (for public nodes). 106hostname, a hostname and a port, or a random string. AnyEvent::MP itself
107doesn't interpret node IDs in any way except to uniquely identify a node.
108
109=item binds - C<ip:port>
110
111Nodes can only talk to each other by creating some kind of connection to
112each other. To do this, nodes should listen on one or more local transport
113endpoints - binds.
114
115Currently, only standard C<ip:port> specifications can be used, which
116specify TCP ports to listen on. So a bind is basically just a tcp socket
117in listening mode thta accepts conenctions form other nodes.
118
119=item seed nodes
120
121When a node starts, it knows nothing about the network it is in - it
122needs to connect to at least one other node that is already in the
123network. These other nodes are called "seed nodes".
124
125Seed nodes themselves are not special - they are seed nodes only because
126some other node I<uses> them as such, but any node can be used as seed
127node for other nodes, and eahc node cna use a different set of seed nodes.
128
129In addition to discovering the network, seed nodes are also used to
130maintain the network - all nodes using the same seed node form are part of
131the same network. If a network is split into multiple subnets because e.g.
132the network link between the parts goes down, then using the same seed
133nodes for all nodes ensures that eventually the subnets get merged again.
134
135Seed nodes are expected to be long-running, and at least one seed node
136should always be available. They should also be relatively responsive - a
137seed node that blocks for long periods will slow down everybody else.
138
139For small networks, it's best if every node uses the same set of seed
140nodes. For large networks, it can be useful to specify "regional" seed
141nodes for most nodes in an area, and use all seed nodes as seed nodes for
142each other. What's important is that all seed nodes connections form a
143complete graph, so that the network cannot split into separate subnets
144forever.
145
146Seed nodes are represented by seed IDs.
147
148=item seed IDs - C<host:port>
149
150Seed IDs are transport endpoint(s) (usually a hostname/IP address and a
151TCP port) of nodes that should be used as seed nodes.
152
153=item global nodes
154
155An AEMP network needs a discovery service - nodes need to know how to
156connect to other nodes they only know by name. In addition, AEMP offers a
157distributed "group database", which maps group names to a list of strings
158- for example, to register worker ports.
159
160A network needs at least one global node to work, and allows every node to
161be a global node.
162
163Any node that loads the L<AnyEvent::MP::Global> module becomes a global
164node and tries to keep connections to all other nodes. So while it can
165make sense to make every node "global" in small networks, it usually makes
166sense to only make seed nodes into global nodes in large networks (nodes
167keep connections to seed nodes and global nodes, so makign them the same
168reduces overhead).
66 169
67=back 170=back
68 171
69=head1 VARIABLES/FUNCTIONS 172=head1 VARIABLES/FUNCTIONS
70 173
72 175
73=cut 176=cut
74 177
75package AnyEvent::MP; 178package AnyEvent::MP;
76 179
77use AnyEvent::MP::Util (); 180use AnyEvent::MP::Config ();
78use AnyEvent::MP::Node; 181use AnyEvent::MP::Kernel;
79use AnyEvent::MP::Transport; 182use AnyEvent::MP::Kernel qw(%NODE %PORT %PORT_DATA $UNIQ $RUNIQ $ID);
80 183
81use utf8;
82use common::sense; 184use common::sense;
83 185
84use Carp (); 186use Carp ();
85 187
86use AE (); 188use AE ();
87 189
88use base "Exporter"; 190use base "Exporter";
89 191
90our $VERSION = '0.01'; 192our $VERSION = $AnyEvent::MP::Config::VERSION;
91our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_);
92 193
93our $DEFAULT_SECRET; 194our @EXPORT = qw(
94our $DEFAULT_PORT = "4040"; 195 NODE $NODE *SELF node_of after
196 configure
197 snd rcv mon mon_guard kil psub peval spawn cal
198 port
199);
95 200
96our $CONNECT_INTERVAL = 5; # new connect every 5s, at least 201our $SELF;
97our $CONNECT_TIMEOUT = 30; # includes handshake
98 202
99sub default_secret { 203sub _self_die() {
100 unless (defined $DEFAULT_SECRET) { 204 my $msg = $@;
101 if (open my $fh, "<$ENV{HOME}/.aemp-secret") { 205 $msg =~ s/\n+$// unless ref $msg;
102 sysread $fh, $DEFAULT_SECRET, -s $fh; 206 kil $SELF, die => $msg;
103 } else {
104 $DEFAULT_SECRET = AnyEvent::MP::Util::nonce 32;
105 }
106 }
107
108 $DEFAULT_SECRET
109} 207}
110 208
111=item NODE / $NODE 209=item $thisnode = NODE / $NODE
112 210
113The C<NODE ()> function and the C<$NODE> variable contain the noderef of 211The C<NODE> function returns, and the C<$NODE> variable contains, the node
114the local node. The value is initialised by a call to C<become_public> or 212ID of the node running in the current process. This value is initialised by
115C<become_slave>, after which all local port identifiers become invalid. 213a call to C<configure>.
116 214
117=cut 215=item $nodeid = node_of $port
118 216
119our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie 217Extracts and returns the node ID from a port ID or a node ID.
120our $ID = "a0";
121our $PUBLIC = 0;
122our $NODE;
123our $PORT;
124 218
125our %NODE; # node id to transport mapping, or "undef", for local node 219=item configure $profile, key => value...
126our %PORT; # local ports
127our %LISTENER; # local transports
128 220
129sub NODE() { $NODE } 221=item configure key => value...
130 222
131{ 223Before a node can talk to other nodes on the network (i.e. enter
132 use POSIX (); 224"distributed mode") it has to configure itself - the minimum a node needs
133 my $nodename = (POSIX::uname)[1]; 225to know is its own name, and optionally it should know the addresses of
134 $NODE = "$$\@$nodename"; 226some other nodes in the network to discover other nodes.
135}
136 227
137sub _ANY_() { 1 } 228This function configures a node - it must be called exactly once (or
138sub _any_() { \&_ANY_ } 229never) before calling other AnyEvent::MP functions.
139 230
140sub add_node { 231The key/value pairs are basically the same ones as documented for the
141 my ($noderef) = @_; 232F<aemp> command line utility (sans the set/del prefix), with two additions:
142 233
143 return $NODE{$noderef} 234=over 4
144 if exists $NODE{$noderef};
145 235
146 for (split /,/, $noderef) { 236=item norc => $boolean (default false)
147 return $NODE{$noderef} = $NODE{$_}
148 if exists $NODE{$_};
149 }
150 237
151 # for indirect sends, use a different class 238If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not>
152 my $node = new AnyEvent::MP::Node::Direct $noderef; 239be consulted - all configuraiton options must be specified in the
240C<configure> call.
153 241
154 $NODE{$_} = $node 242=item force => $boolean (default false)
155 for $noderef, split /,/, $noderef;
156 243
157 $node 244IF true, then the values specified in the C<configure> will take
158} 245precedence over any values configured via the rc file. The default is for
246the rc file to override any options specified in the program.
159 247
248=back
249
250=over 4
251
252=item step 1, gathering configuration from profiles
253
254The function first looks up a profile in the aemp configuration (see the
255L<aemp> commandline utility). The profile name can be specified via the
256named C<profile> parameter or can simply be the first parameter). If it is
257missing, then the nodename (F<uname -n>) will be used as profile name.
258
259The profile data is then gathered as follows:
260
261First, all remaining key => value pairs (all of which are conveniently
262undocumented at the moment) will be interpreted as configuration
263data. Then they will be overwritten by any values specified in the global
264default configuration (see the F<aemp> utility), then the chain of
265profiles chosen by the profile name (and any C<parent> attributes).
266
267That means that the values specified in the profile have highest priority
268and the values specified directly via C<configure> have lowest priority,
269and can only be used to specify defaults.
270
271If the profile specifies a node ID, then this will become the node ID of
272this process. If not, then the profile name will be used as node ID, with
273a slash (C</>) attached.
274
275If the node ID (or profile name) ends with a slash (C</>), then a random
276string is appended to make it unique.
277
278=item step 2, bind listener sockets
279
280The next step is to look up the binds in the profile, followed by binding
281aemp protocol listeners on all binds specified (it is possible and valid
282to have no binds, meaning that the node cannot be contacted form the
283outside. This means the node cannot talk to other nodes that also have no
284binds, but it can still talk to all "normal" nodes).
285
286If the profile does not specify a binds list, then a default of C<*> is
287used, meaning the node will bind on a dynamically-assigned port on every
288local IP address it finds.
289
290=item step 3, connect to seed nodes
291
292As the last step, the seed ID list from the profile is passed to the
293L<AnyEvent::MP::Global> module, which will then use it to keep
294connectivity with at least one node at any point in time.
295
296=back
297
298Example: become a distributed node using the local node name as profile.
299This should be the most common form of invocation for "daemon"-type nodes.
300
301 configure
302
303Example: become an anonymous node. This form is often used for commandline
304clients.
305
306 configure nodeid => "anon/";
307
308Example: configure a node using a profile called seed, which is suitable
309for a seed node as it binds on all local addresses on a fixed port (4040,
310customary for aemp).
311
312 # use the aemp commandline utility
313 # aemp profile seed binds '*:4040'
314
315 # then use it
316 configure profile => "seed";
317
318 # or simply use aemp from the shell again:
319 # aemp run profile seed
320
321 # or provide a nicer-to-remember nodeid
322 # aemp run profile seed nodeid "$(hostname)"
323
324=item $SELF
325
326Contains the current port id while executing C<rcv> callbacks or C<psub>
327blocks.
328
329=item *SELF, SELF, %SELF, @SELF...
330
331Due to some quirks in how perl exports variables, it is impossible to
332just export C<$SELF>, all the symbols named C<SELF> are exported by this
333module, but only C<$SELF> is currently used.
334
160=item snd $portid, type => @data 335=item snd $port, type => @data
161 336
162=item snd $portid, @msg 337=item snd $port, @msg
163 338
164Send the given message to the given port ID, which can identify either a 339Send the given message to the given port, which can identify either a
165local or a remote port. 340local or a remote port, and must be a port ID.
166 341
167While the message can be about anything, it is highly recommended to use 342While the message can be almost anything, it is highly recommended to
168a constant string as first element. 343use a string as first element (a port ID, or some word that indicates a
344request type etc.) and to consist if only simple perl values (scalars,
345arrays, hashes) - if you think you need to pass an object, think again.
169 346
170The message data effectively becomes read-only after a call to this 347The message data logically becomes read-only after a call to this
171function: modifying any argument is not allowed and can cause many 348function: modifying any argument (or values referenced by them) is
172problems. 349forbidden, as there can be considerable time between the call to C<snd>
350and the time the message is actually being serialised - in fact, it might
351never be copied as within the same process it is simply handed to the
352receiving port.
173 353
174The type of data you can transfer depends on the transport protocol: when 354The type of data you can transfer depends on the transport protocol: when
175JSON is used, then only strings, numbers and arrays and hashes consisting 355JSON is used, then only strings, numbers and arrays and hashes consisting
176of those are allowed (no objects). When Storable is used, then anything 356of those are allowed (no objects). When Storable is used, then anything
177that Storable can serialise and deserialise is allowed, and for the local 357that Storable can serialise and deserialise is allowed, and for the local
178node, anything can be passed. 358node, anything can be passed. Best rely only on the common denominator of
359these.
179 360
180=cut 361=item $local_port = port
181 362
182sub snd(@) { 363Create a new local port object and returns its port ID. Initially it has
183 my ($noderef, $port) = split /#/, shift, 2; 364no callbacks set and will throw an error when it receives messages.
184 365
185 add_node $noderef 366=item $local_port = port { my @msg = @_ }
186 unless exists $NODE{$noderef};
187 367
188 $NODE{$noderef}->send (["$port", [@_]]); 368Creates a new local port, and returns its ID. Semantically the same as
189} 369creating a port and calling C<rcv $port, $callback> on it.
190 370
371The block will be called for every message received on the port, with the
372global variable C<$SELF> set to the port ID. Runtime errors will cause the
373port to be C<kil>ed. The message will be passed as-is, no extra argument
374(i.e. no port ID) will be passed to the callback.
375
376If you want to stop/destroy the port, simply C<kil> it:
377
378 my $port = port {
379 my @msg = @_;
380 ...
381 kil $SELF;
382 };
383
384=cut
385
386sub rcv($@);
387
388sub _kilme {
389 die "received message on port without callback";
390}
391
392sub port(;&) {
393 my $id = "$UNIQ." . ++$ID;
394 my $port = "$NODE#$id";
395
396 rcv $port, shift || \&_kilme;
397
398 $port
399}
400
191=item rcv $portid, type => $callback->(@msg) 401=item rcv $local_port, $callback->(@msg)
192 402
193=item rcv $portid, $smartmatch => $callback->(@msg) 403Replaces the default callback on the specified port. There is no way to
404remove the default callback: use C<sub { }> to disable it, or better
405C<kil> the port when it is no longer needed.
194 406
195=item rcv $portid, [$smartmatch...] => $callback->(@msg) 407The global C<$SELF> (exported by this module) contains C<$port> while
408executing the callback. Runtime errors during callback execution will
409result in the port being C<kil>ed.
196 410
197Register a callback on the port identified by C<$portid>, which I<must> be 411The default callback received all messages not matched by a more specific
198a local port. 412C<tag> match.
199 413
200The callback has to return a true value when its work is done, after 414=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
201which is will be removed, or a false value in which case it will stay
202registered.
203 415
204If the match is an array reference, then it will be matched against the 416Register (or replace) callbacks to be called on messages starting with the
205first elements of the message, otherwise only the first element is being 417given tag on the given port (and return the port), or unregister it (when
206matched. 418C<$callback> is C<$undef> or missing). There can only be one callback
419registered for each tag.
207 420
208Any element in the match that is specified as C<_any_> (a function 421The original message will be passed to the callback, after the first
209exported by this module) matches any single element of the message. 422element (the tag) has been removed. The callback will use the same
423environment as the default callback (see above).
210 424
211While not required, it is highly recommended that the first matching 425Example: create a port and bind receivers on it in one go.
212element is a string identifying the message. The one-string-only match is 426
213also the most efficient match (by far). 427 my $port = rcv port,
428 msg1 => sub { ... },
429 msg2 => sub { ... },
430 ;
431
432Example: create a port, bind receivers and send it in a message elsewhere
433in one go:
434
435 snd $otherport, reply =>
436 rcv port,
437 msg1 => sub { ... },
438 ...
439 ;
440
441Example: temporarily register a rcv callback for a tag matching some port
442(e.g. for an rpc reply) and unregister it after a message was received.
443
444 rcv $port, $otherport => sub {
445 my @reply = @_;
446
447 rcv $SELF, $otherport;
448 };
214 449
215=cut 450=cut
216 451
217sub rcv($@) { 452sub rcv($@) {
218 my ($port, $match, $cb) = @_; 453 my $port = shift;
219
220 my $port = $PORT{$port}
221 or do {
222 my ($noderef, $lport) = split /#/, $port; 454 my ($nodeid, $portid) = split /#/, $port, 2;
223 "AnyEvent::MP::Node::Self" eq ref $NODE{$noderef} 455
456 $NODE{$nodeid} == $NODE{""}
224 or Carp::croak "$port: can only rcv on local ports"; 457 or Carp::croak "$port: rcv can only be called on local ports, caught";
225 458
226 $PORT{$lport} 459 while (@_) {
227 or Carp::croak "$port: port does not exist"; 460 if (ref $_[0]) {
228 461 if (my $self = $PORT_DATA{$portid}) {
229 $PORT{$port} = $PORT{$lport} # also return 462 "AnyEvent::MP::Port" eq ref $self
230 }; 463 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
231 464
232 if (!ref $match) { 465 $self->[0] = shift;
233 push @{ $port->{rc0}{$match} }, [$cb];
234 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
235 my ($type, @match) = @$match;
236 @match
237 ? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match]
238 : push @{ $port->{rc0}{$match->[0]} }, [$cb];
239 } else { 466 } else {
240 push @{ $port->{any} }, [$cb, $match]; 467 my $cb = shift;
241 } 468 $PORT{$portid} = sub {
242} 469 local $SELF = $port;
243 470 eval { &$cb }; _self_die if $@;
244sub _inject {
245 my ($port, $msg) = @{+shift};
246
247 $port = $PORT{$port}
248 or return;
249
250 @_ = @$msg;
251
252 for (@{ $port->{rc0}{$msg->[0]} }) {
253 $_ && &{$_->[0]}
254 && undef $_;
255 }
256
257 for (@{ $port->{rcv}{$msg->[0]} }) {
258 $_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1]
259 && &{$_->[0]}
260 && undef $_;
261 }
262
263 for (@{ $port->{any} }) {
264 $_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1]
265 && &{$_->[0]}
266 && undef $_;
267 }
268}
269
270sub normalise_noderef($) {
271 my ($noderef) = @_;
272
273 my $cv = AE::cv;
274 my @res;
275
276 $cv->begin (sub {
277 my %seen;
278 my @refs;
279 for (sort { $a->[0] <=> $b->[0] } @res) {
280 push @refs, $_->[1] unless $seen{$_->[1]}++
281 }
282 shift->send (join ",", @refs);
283 });
284
285 $noderef = $DEFAULT_PORT unless length $noderef;
286
287 my $idx;
288 for my $t (split /,/, $noderef) {
289 my $pri = ++$idx;
290
291 #TODO: this should be outside normalise_noderef and in become_public
292 if ($t =~ /^\d*$/) {
293 my $nodename = (POSIX::uname)[1];
294
295 $cv->begin;
296 AnyEvent::Socket::resolve_sockaddr $nodename, $t || "aemp=$DEFAULT_PORT", "tcp", 0, undef, sub {
297 for (@_) {
298 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
299 push @res, [
300 $pri += 1e-5,
301 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
302 ];
303 } 471 };
304 $cv->end; 472 }
473 } elsif (defined $_[0]) {
474 my $self = $PORT_DATA{$portid} ||= do {
475 my $self = bless [$PORT{$portid} || sub { }, { }, $port], "AnyEvent::MP::Port";
476
477 $PORT{$portid} = sub {
478 local $SELF = $port;
479
480 if (my $cb = $self->[1]{$_[0]}) {
481 shift;
482 eval { &$cb }; _self_die if $@;
483 } else {
484 &{ $self->[0] };
485 }
486 };
487
488 $self
305 }; 489 };
306 490
307# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename; 491 "AnyEvent::MP::Port" eq ref $self
308# 492 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
309# for (@ipv4) { 493
310# push @res, [ 494 my ($tag, $cb) = splice @_, 0, 2;
311# $pri, 495
312# AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $_, $t || $DEFAULT_PORT, 496 if (defined $cb) {
313# ]; 497 $self->[1]{$tag} = $cb;
314# }
315 } else { 498 } else {
316 my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT" 499 delete $self->[1]{$tag};
317 or Carp::croak "$t: unparsable transport descriptor";
318
319 $cv->begin;
320 AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {
321 for (@_) {
322 my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
323 push @res, [
324 $pri += 1e-5,
325 AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
326 ];
327 }
328 $cv->end;
329 } 500 }
330 } 501 }
331 } 502 }
332 503
333 $cv->end; 504 $port
334
335 $cv
336} 505}
337 506
338sub become_public { 507=item peval $port, $coderef[, @args]
339 return if $PUBLIC;
340 508
341 my $noderef = join ",", ref $_[0] ? @{+shift} : shift; 509Evaluates the given C<$codref> within the contetx of C<$port>, that is,
342 my @args = @_; 510when the code throews an exception the C<$port> will be killed.
343 511
344 $NODE = (normalise_noderef $noderef)->recv; 512Any remaining args will be passed to the callback. Any return values will
513be returned to the caller.
345 514
346 for my $t (split /,/, $NODE) { 515This is useful when you temporarily want to execute code in the context of
347 $NODE{$t} = $NODE{""}; 516a port.
348 517
349 my ($host, $port) = AnyEvent::Socket::parse_hostport $t; 518Example: create a port and run some initialisation code in it's context.
350 519
351 $LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port, 520 my $port = port { ... };
352 @args,
353 on_error => sub {
354 die "on_error<@_>\n";#d#
355 },
356 on_connect => sub {
357 my ($tp) = @_;
358 521
359 $NODE{$tp->{remote_id}} = $_[0]; 522 peval $port, sub {
360 }, 523 init
361 sub { 524 or die "unable to init";
362 my ($tp) = @_; 525 };
363 526
364 $NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp; 527=cut
365 }, 528
366 ; 529sub peval($$) {
530 local $SELF = shift;
531 my $cb = shift;
532
533 if (wantarray) {
534 my @res = eval { &$cb };
535 _self_die if $@;
536 @res
537 } else {
538 my $res = eval { &$cb };
539 _self_die if $@;
540 $res
367 } 541 }
542}
368 543
369 $PUBLIC = 1; 544=item $closure = psub { BLOCK }
545
546Remembers C<$SELF> and creates a closure out of the BLOCK. When the
547closure is executed, sets up the environment in the same way as in C<rcv>
548callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
549
550The effect is basically as if it returned C<< sub { peval $SELF, sub {
551BLOCK }, @_ } >>.
552
553This is useful when you register callbacks from C<rcv> callbacks:
554
555 rcv delayed_reply => sub {
556 my ($delay, @reply) = @_;
557 my $timer = AE::timer $delay, 0, psub {
558 snd @reply, $SELF;
559 };
560 };
561
562=cut
563
564sub psub(&) {
565 my $cb = shift;
566
567 my $port = $SELF
568 or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
569
570 sub {
571 local $SELF = $port;
572
573 if (wantarray) {
574 my @res = eval { &$cb };
575 _self_die if $@;
576 @res
577 } else {
578 my $res = eval { &$cb };
579 _self_die if $@;
580 $res
581 }
582 }
583}
584
585=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
586
587=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
588
589=item $guard = mon $port # kill $SELF when $port dies
590
591=item $guard = mon $port, $rcvport, @msg # send a message when $port dies
592
593Monitor the given port and do something when the port is killed or
594messages to it were lost, and optionally return a guard that can be used
595to stop monitoring again.
596
597In the first form (callback), the callback is simply called with any
598number of C<@reason> elements (no @reason means that the port was deleted
599"normally"). Note also that I<< the callback B<must> never die >>, so use
600C<eval> if unsure.
601
602In the second form (another port given), the other port (C<$rcvport>)
603will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
604"normal" kils nothing happens, while under all other conditions, the other
605port is killed with the same reason.
606
607The third form (kill self) is the same as the second form, except that
608C<$rvport> defaults to C<$SELF>.
609
610In the last form (message), a message of the form C<@msg, @reason> will be
611C<snd>.
612
613Monitoring-actions are one-shot: once messages are lost (and a monitoring
614alert was raised), they are removed and will not trigger again.
615
616As a rule of thumb, monitoring requests should always monitor a port from
617a local port (or callback). The reason is that kill messages might get
618lost, just like any other message. Another less obvious reason is that
619even monitoring requests can get lost (for example, when the connection
620to the other node goes down permanently). When monitoring a port locally
621these problems do not exist.
622
623C<mon> effectively guarantees that, in the absence of hardware failures,
624after starting the monitor, either all messages sent to the port will
625arrive, or the monitoring action will be invoked after possible message
626loss has been detected. No messages will be lost "in between" (after
627the first lost message no further messages will be received by the
628port). After the monitoring action was invoked, further messages might get
629delivered again.
630
631Inter-host-connection timeouts and monitoring depend on the transport
632used. The only transport currently implemented is TCP, and AnyEvent::MP
633relies on TCP to detect node-downs (this can take 10-15 minutes on a
634non-idle connection, and usually around two hours for idle connections).
635
636This means that monitoring is good for program errors and cleaning up
637stuff eventually, but they are no replacement for a timeout when you need
638to ensure some maximum latency.
639
640Example: call a given callback when C<$port> is killed.
641
642 mon $port, sub { warn "port died because of <@_>\n" };
643
644Example: kill ourselves when C<$port> is killed abnormally.
645
646 mon $port;
647
648Example: send us a restart message when another C<$port> is killed.
649
650 mon $port, $self => "restart";
651
652=cut
653
654sub mon {
655 my ($nodeid, $port) = split /#/, shift, 2;
656
657 my $node = $NODE{$nodeid} || add_node $nodeid;
658
659 my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,';
660
661 unless (ref $cb) {
662 if (@_) {
663 # send a kill info message
664 my (@msg) = ($cb, @_);
665 $cb = sub { snd @msg, @_ };
666 } else {
667 # simply kill other port
668 my $port = $cb;
669 $cb = sub { kil $port, @_ if @_ };
670 }
671 }
672
673 $node->monitor ($port, $cb);
674
675 defined wantarray
676 and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) })
677}
678
679=item $guard = mon_guard $port, $ref, $ref...
680
681Monitors the given C<$port> and keeps the passed references. When the port
682is killed, the references will be freed.
683
684Optionally returns a guard that will stop the monitoring.
685
686This function is useful when you create e.g. timers or other watchers and
687want to free them when the port gets killed (note the use of C<psub>):
688
689 $port->rcv (start => sub {
690 my $timer; $timer = mon_guard $port, AE::timer 1, 1, psub {
691 undef $timer if 0.9 < rand;
692 });
693 });
694
695=cut
696
697sub mon_guard {
698 my ($port, @refs) = @_;
699
700 #TODO: mon-less form?
701
702 mon $port, sub { 0 && @refs }
703}
704
705=item kil $port[, @reason]
706
707Kill the specified port with the given C<@reason>.
708
709If no C<@reason> is specified, then the port is killed "normally" -
710monitor callback will be invoked, but the kil will not cause linked ports
711(C<mon $mport, $lport> form) to get killed.
712
713If a C<@reason> is specified, then linked ports (C<mon $mport, $lport>
714form) get killed with the same reason.
715
716Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
717will be reported as reason C<< die => $@ >>.
718
719Transport/communication errors are reported as C<< transport_error =>
720$message >>.
721
722=cut
723
724=item $port = spawn $node, $initfunc[, @initdata]
725
726Creates a port on the node C<$node> (which can also be a port ID, in which
727case it's the node where that port resides).
728
729The port ID of the newly created port is returned immediately, and it is
730possible to immediately start sending messages or to monitor the port.
731
732After the port has been created, the init function is called on the remote
733node, in the same context as a C<rcv> callback. This function must be a
734fully-qualified function name (e.g. C<MyApp::Chat::Server::init>). To
735specify a function in the main program, use C<::name>.
736
737If the function doesn't exist, then the node tries to C<require>
738the package, then the package above the package and so on (e.g.
739C<MyApp::Chat::Server>, C<MyApp::Chat>, C<MyApp>) until the function
740exists or it runs out of package names.
741
742The init function is then called with the newly-created port as context
743object (C<$SELF>) and the C<@initdata> values as arguments. It I<must>
744call one of the C<rcv> functions to set callbacks on C<$SELF>, otherwise
745the port might not get created.
746
747A common idiom is to pass a local port, immediately monitor the spawned
748port, and in the remote init function, immediately monitor the passed
749local port. This two-way monitoring ensures that both ports get cleaned up
750when there is a problem.
751
752C<spawn> guarantees that the C<$initfunc> has no visible effects on the
753caller before C<spawn> returns (by delaying invocation when spawn is
754called for the local node).
755
756Example: spawn a chat server port on C<$othernode>.
757
758 # this node, executed from within a port context:
759 my $server = spawn $othernode, "MyApp::Chat::Server::connect", $SELF;
760 mon $server;
761
762 # init function on C<$othernode>
763 sub connect {
764 my ($srcport) = @_;
765
766 mon $srcport;
767
768 rcv $SELF, sub {
769 ...
770 };
771 }
772
773=cut
774
775sub _spawn {
776 my $port = shift;
777 my $init = shift;
778
779 # rcv will create the actual port
780 local $SELF = "$NODE#$port";
781 eval {
782 &{ load_func $init }
783 };
784 _self_die if $@;
785}
786
787sub spawn(@) {
788 my ($nodeid, undef) = split /#/, shift, 2;
789
790 my $id = "$RUNIQ." . ++$ID;
791
792 $_[0] =~ /::/
793 or Carp::croak "spawn init function must be a fully-qualified name, caught";
794
795 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
796
797 "$nodeid#$id"
798}
799
800
801=item after $timeout, @msg
802
803=item after $timeout, $callback
804
805Either sends the given message, or call the given callback, after the
806specified number of seconds.
807
808This is simply a utility function that comes in handy at times - the
809AnyEvent::MP author is not convinced of the wisdom of having it, though,
810so it may go away in the future.
811
812=cut
813
814sub after($@) {
815 my ($timeout, @action) = @_;
816
817 my $t; $t = AE::timer $timeout, 0, sub {
818 undef $t;
819 ref $action[0]
820 ? $action[0]()
821 : snd @action;
822 };
823}
824
825=item cal $port, @msg, $callback[, $timeout]
826
827A simple form of RPC - sends a message to the given C<$port> with the
828given contents (C<@msg>), but adds a reply port to the message.
829
830The reply port is created temporarily just for the purpose of receiving
831the reply, and will be C<kil>ed when no longer needed.
832
833A reply message sent to the port is passed to the C<$callback> as-is.
834
835If an optional time-out (in seconds) is given and it is not C<undef>,
836then the callback will be called without any arguments after the time-out
837elapsed and the port is C<kil>ed.
838
839If no time-out is given (or it is C<undef>), then the local port will
840monitor the remote port instead, so it eventually gets cleaned-up.
841
842Currently this function returns the temporary port, but this "feature"
843might go in future versions unless you can make a convincing case that
844this is indeed useful for something.
845
846=cut
847
848sub cal(@) {
849 my $timeout = ref $_[-1] ? undef : pop;
850 my $cb = pop;
851
852 my $port = port {
853 undef $timeout;
854 kil $SELF;
855 &$cb;
856 };
857
858 if (defined $timeout) {
859 $timeout = AE::timer $timeout, 0, sub {
860 undef $timeout;
861 kil $port;
862 $cb->();
863 };
864 } else {
865 mon $_[0], sub {
866 kil $port;
867 $cb->();
868 };
869 }
870
871 push @_, $port;
872 &snd;
873
874 $port
370} 875}
371 876
372=back 877=back
373 878
374=head1 NODE MESSAGES 879=head1 AnyEvent::MP vs. Distributed Erlang
375 880
376Nodes understand the following messages sent to them. Many of them take 881AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
377arguments called C<@reply>, which will simply be used to compose a reply 882== aemp node, Erlang process == aemp port), so many of the documents and
378message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and 883programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
379the remaining arguments are simply the message data. 884sample:
885
886 http://www.erlang.se/doc/programming_rules.shtml
887 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
888 http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6
889 http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
890
891Despite the similarities, there are also some important differences:
380 892
381=over 4 893=over 4
382 894
383=cut 895=item * Node IDs are arbitrary strings in AEMP.
384 896
385############################################################################# 897Erlang relies on special naming and DNS to work everywhere in the same
386# self node code 898way. AEMP relies on each node somehow knowing its own address(es) (e.g. by
899configuration or DNS), and possibly the addresses of some seed nodes, but
900will otherwise discover other nodes (and their IDs) itself.
387 901
388sub _new_port($) { 902=item * Erlang has a "remote ports are like local ports" philosophy, AEMP
389 my ($name) = @_; 903uses "local ports are like remote ports".
390 904
391 my ($noderef, $portname) = split /#/, $name; 905The failure modes for local ports are quite different (runtime errors
906only) then for remote ports - when a local port dies, you I<know> it dies,
907when a connection to another node dies, you know nothing about the other
908port.
392 909
393 $PORT{$name} = 910Erlang pretends remote ports are as reliable as local ports, even when
394 $PORT{$portname} = { 911they are not.
395 names => [$name, $portname],
396 };
397}
398 912
399$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE; 913AEMP encourages a "treat remote ports differently" philosophy, with local
400_new_port ""; 914ports being the special case/exception, where transport errors cannot
915occur.
401 916
402=item devnull => ... 917=item * Erlang uses processes and a mailbox, AEMP does not queue.
403 918
404Generic data sink/CPU heat conversion. 919Erlang uses processes that selectively receive messages out of order, and
920therefore needs a queue. AEMP is event based, queuing messages would serve
921no useful purpose. For the same reason the pattern-matching abilities
922of AnyEvent::MP are more limited, as there is little need to be able to
923filter messages without dequeuing them.
405 924
406=cut 925This is not a philosophical difference, but simply stems from AnyEvent::MP
926being event-based, while Erlang is process-based.
407 927
408rcv "", devnull => sub { () }; 928You cna have a look at L<Coro::MP> for a more Erlang-like process model on
929top of AEMP and Coro threads.
409 930
410=item relay => $port, @msg 931=item * Erlang sends are synchronous, AEMP sends are asynchronous.
411 932
412Simply forwards the message to the given port. 933Sending messages in Erlang is synchronous and blocks the process until
934a conenction has been established and the message sent (and so does not
935need a queue that can overflow). AEMP sends return immediately, connection
936establishment is handled in the background.
413 937
414=cut 938=item * Erlang suffers from silent message loss, AEMP does not.
415 939
416rcv "", relay => sub { \&snd; () }; 940Erlang implements few guarantees on messages delivery - messages can get
941lost without any of the processes realising it (i.e. you send messages a,
942b, and c, and the other side only receives messages a and c).
417 943
418=item eval => $string[ @reply] 944AEMP guarantees (modulo hardware errors) correct ordering, and the
945guarantee that after one message is lost, all following ones sent to the
946same port are lost as well, until monitoring raises an error, so there are
947no silent "holes" in the message sequence.
419 948
420Evaluates the given string. If C<@reply> is given, then a message of the 949If you want your software to be very reliable, you have to cope with
421form C<@reply, $@, @evalres> is sent. 950corrupted and even out-of-order messages in both Erlang and AEMP. AEMP
951simply tries to work better in common error cases, such as when a network
952link goes down.
422 953
423Example: crash another node. 954=item * Erlang can send messages to the wrong port, AEMP does not.
424 955
425 snd $othernode, eval => "exit"; 956In Erlang it is quite likely that a node that restarts reuses an Erlang
957process ID known to other nodes for a completely different process,
958causing messages destined for that process to end up in an unrelated
959process.
426 960
427=cut 961AEMP does not reuse port IDs, so old messages or old port IDs floating
962around in the network will not be sent to an unrelated port.
428 963
429rcv "", eval => sub { 964=item * Erlang uses unprotected connections, AEMP uses secure
430 my (undef, $string, @reply) = @_; 965authentication and can use TLS.
431 my @res = eval $string;
432 snd @reply, "$@", @res if @reply;
433 ()
434};
435 966
436=item time => @reply 967AEMP can use a proven protocol - TLS - to protect connections and
968securely authenticate nodes.
437 969
438Replies the the current node time to C<@reply>. 970=item * The AEMP protocol is optimised for both text-based and binary
971communications.
439 972
440Example: tell the current node to send the current time to C<$myport> in a 973The AEMP protocol, unlike the Erlang protocol, supports both programming
441C<timereply> message. 974language independent text-only protocols (good for debugging), and binary,
975language-specific serialisers (e.g. Storable). By default, unless TLS is
976used, the protocol is actually completely text-based.
442 977
443 snd $NODE, time => $myport, timereply => 1, 2; 978It has also been carefully designed to be implementable in other languages
444 # => snd $myport, timereply => 1, 2, <time> 979with a minimum of work while gracefully degrading functionality to make the
980protocol simple.
445 981
446=cut 982=item * AEMP has more flexible monitoring options than Erlang.
447 983
448rcv "", time => sub { shift; snd @_, AE::time; () }; 984In Erlang, you can chose to receive I<all> exit signals as messages or
985I<none>, there is no in-between, so monitoring single Erlang processes is
986difficult to implement.
987
988Monitoring in AEMP is more flexible than in Erlang, as one can choose
989between automatic kill, exit message or callback on a per-port basis.
990
991=item * Erlang tries to hide remote/local connections, AEMP does not.
992
993Monitoring in Erlang is not an indicator of process death/crashes, in the
994same way as linking is (except linking is unreliable in Erlang).
995
996In AEMP, you don't "look up" registered port names or send to named ports
997that might or might not be persistent. Instead, you normally spawn a port
998on the remote node. The init function monitors you, and you monitor the
999remote port. Since both monitors are local to the node, they are much more
1000reliable (no need for C<spawn_link>).
1001
1002This also saves round-trips and avoids sending messages to the wrong port
1003(hard to do in Erlang).
449 1004
450=back 1005=back
451 1006
1007=head1 RATIONALE
1008
1009=over 4
1010
1011=item Why strings for port and node IDs, why not objects?
1012
1013We considered "objects", but found that the actual number of methods
1014that can be called are quite low. Since port and node IDs travel over
1015the network frequently, the serialising/deserialising would add lots of
1016overhead, as well as having to keep a proxy object everywhere.
1017
1018Strings can easily be printed, easily serialised etc. and need no special
1019procedures to be "valid".
1020
1021And as a result, a port with just a default receiver consists of a single
1022code reference stored in a global hash - it can't become much cheaper.
1023
1024=item Why favour JSON, why not a real serialising format such as Storable?
1025
1026In fact, any AnyEvent::MP node will happily accept Storable as framing
1027format, but currently there is no way to make a node use Storable by
1028default (although all nodes will accept it).
1029
1030The default framing protocol is JSON because a) JSON::XS is many times
1031faster for small messages and b) most importantly, after years of
1032experience we found that object serialisation is causing more problems
1033than it solves: Just like function calls, objects simply do not travel
1034easily over the network, mostly because they will always be a copy, so you
1035always have to re-think your design.
1036
1037Keeping your messages simple, concentrating on data structures rather than
1038objects, will keep your messages clean, tidy and efficient.
1039
1040=back
1041
452=head1 SEE ALSO 1042=head1 SEE ALSO
1043
1044L<AnyEvent::MP::Intro> - a gentle introduction.
1045
1046L<AnyEvent::MP::Kernel> - more, lower-level, stuff.
1047
1048L<AnyEvent::MP::Global> - network maintenance and port groups, to find
1049your applications.
1050
1051L<AnyEvent::MP::DataConn> - establish data connections between nodes.
1052
1053L<AnyEvent::MP::LogCatcher> - simple service to display log messages from
1054all nodes.
453 1055
454L<AnyEvent>. 1056L<AnyEvent>.
455 1057
456=head1 AUTHOR 1058=head1 AUTHOR
457 1059

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines