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.121 by root, Tue Feb 28 18:37:24 2012 UTC vs.
Revision 1.137 by root, Wed Mar 21 23:48:39 2012 UTC

35 # destroy a port again 35 # destroy a port again
36 kil $port; # "normal" kill 36 kil $port; # "normal" kill
37 kil $port, my_error => "everything is broken"; # error kill 37 kil $port, my_error => "everything is broken"; # error kill
38 38
39 # monitoring 39 # monitoring
40 mon $localport, $cb->(@msg) # callback is invoked on death 40 mon $port, $cb->(@msg) # callback is invoked on death
41 mon $localport, $otherport # kill otherport on abnormal death 41 mon $port, $localport # kill localport on abnormal death
42 mon $localport, $otherport, @msg # send message on death 42 mon $port, $localport, @msg # send message on death
43 43
44 # temporarily execute code in port context 44 # temporarily execute code in port context
45 peval $port, sub { die "kill the port!" }; 45 peval $port, sub { die "kill the port!" };
46 46
47 # execute callbacks in $SELF port context 47 # execute callbacks in $SELF port context
82 82
83Ports are represented by (printable) strings called "port IDs". 83Ports are represented by (printable) strings called "port IDs".
84 84
85=item port ID - C<nodeid#portname> 85=item port ID - C<nodeid#portname>
86 86
87A port ID is the concatenation of a node ID, a hash-mark (C<#>) as 87A port ID is the concatenation of a node ID, a hash-mark (C<#>)
88separator, and a port name (a printable string of unspecified format). 88as separator, and a port name (a printable string of unspecified
89format created by AnyEvent::MP).
89 90
90=item node 91=item node
91 92
92A node is a single process containing at least one port - the node port, 93A node is a single process containing at least one port - the node port,
93which enables nodes to manage each other remotely, and to create new 94which enables nodes to manage each other remotely, and to create new
184use common::sense; 185use common::sense;
185 186
186use Carp (); 187use Carp ();
187 188
188use AE (); 189use AE ();
190use Guard ();
189 191
190use base "Exporter"; 192use base "Exporter";
191 193
192our $VERSION = $AnyEvent::MP::Config::VERSION; 194our $VERSION = $AnyEvent::MP::Config::VERSION;
193 195
194our @EXPORT = qw( 196our @EXPORT = qw(
195 NODE $NODE *SELF node_of after 197 NODE $NODE *SELF node_of after
196 configure 198 configure
197 snd rcv mon mon_guard kil psub peval spawn cal 199 snd rcv mon mon_guard kil psub peval spawn cal
198 port 200 port
201 db_set db_del db_reg
202 db_mon db_family db_keys db_values
199); 203);
200 204
201our $SELF; 205our $SELF;
202 206
203sub _self_die() { 207sub _self_die() {
227 231
228This function configures a node - it must be called exactly once (or 232This function configures a node - it must be called exactly once (or
229never) before calling other AnyEvent::MP functions. 233never) before calling other AnyEvent::MP functions.
230 234
231The key/value pairs are basically the same ones as documented for the 235The key/value pairs are basically the same ones as documented for the
232F<aemp> command line utility (sans the set/del prefix), with two additions: 236F<aemp> command line utility (sans the set/del prefix), with these additions:
233 237
234=over 4 238=over 4
235 239
236=item norc => $boolean (default false) 240=item norc => $boolean (default false)
237 241
242=item force => $boolean (default false) 246=item force => $boolean (default false)
243 247
244IF true, then the values specified in the C<configure> will take 248IF true, then the values specified in the C<configure> will take
245precedence over any values configured via the rc file. The default is for 249precedence over any values configured via the rc file. The default is for
246the rc file to override any options specified in the program. 250the rc file to override any options specified in the program.
251
252=item secure => $pass->(@msg)
253
254In addition to specifying a boolean, you can specify a code reference that
255is called for every code execution attempt - the execution request is
256granted iff the callback returns a true value.
257
258Most of the time the callback should look only at
259C<$AnyEvent::MP::Kernel::SRCNODE> to make a decision, and not at the
260actual message (which can be about anything, and is mostly provided for
261diagnostic purposes).
262
263See F<semp setsecure> for more info.
247 264
248=back 265=back
249 266
250=over 4 267=over 4
251 268
267That means that the values specified in the profile have highest priority 284That means that the values specified in the profile have highest priority
268and the values specified directly via C<configure> have lowest priority, 285and the values specified directly via C<configure> have lowest priority,
269and can only be used to specify defaults. 286and can only be used to specify defaults.
270 287
271If the profile specifies a node ID, then this will become the node ID of 288If 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. The 289this process. If not, then the profile name will be used as node ID, with
273special node ID of C<anon/> will be replaced by a random node ID. 290a unique randoms tring (C</%u>) appended.
291
292The node ID can contain some C<%> sequences that are expanded: C<%n>
293is expanded to the local nodename, C<%u> is replaced by a random
294strign to make the node unique. For example, the F<aemp> commandline
295utility uses C<aemp/%n/%u> as nodename, which might expand to
296C<aemp/cerebro/ZQDGSIkRhEZQDGSIkRhE>.
274 297
275=item step 2, bind listener sockets 298=item step 2, bind listener sockets
276 299
277The next step is to look up the binds in the profile, followed by binding 300The next step is to look up the binds in the profile, followed by binding
278aemp protocol listeners on all binds specified (it is possible and valid 301aemp protocol listeners on all binds specified (it is possible and valid
295Example: become a distributed node using the local node name as profile. 318Example: become a distributed node using the local node name as profile.
296This should be the most common form of invocation for "daemon"-type nodes. 319This should be the most common form of invocation for "daemon"-type nodes.
297 320
298 configure 321 configure
299 322
300Example: become an anonymous node. This form is often used for commandline 323Example: become a semi-anonymous node. This form is often used for
301clients. 324commandline clients.
302 325
303 configure nodeid => "anon/"; 326 configure nodeid => "myscript/%n/%u";
304 327
305Example: configure a node using a profile called seed, which is suitable 328Example: configure a node using a profile called seed, which is suitable
306for a seed node as it binds on all local addresses on a fixed port (4040, 329for a seed node as it binds on all local addresses on a fixed port (4040,
307customary for aemp). 330customary for aemp).
308 331
309 # use the aemp commandline utility 332 # use the aemp commandline utility
310 # aemp profile seed nodeid anon/ binds '*:4040' 333 # aemp profile seed binds '*:4040'
311 334
312 # then use it 335 # then use it
313 configure profile => "seed"; 336 configure profile => "seed";
314 337
315 # or simply use aemp from the shell again: 338 # or simply use aemp from the shell again:
380 403
381=cut 404=cut
382 405
383sub rcv($@); 406sub rcv($@);
384 407
385sub _kilme { 408my $KILME = sub {
386 die "received message on port without callback"; 409 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
387} 410 kil $SELF, unhandled_message => "no callback found for message '$tag'";
411};
388 412
389sub port(;&) { 413sub port(;&) {
390 my $id = "$UNIQ." . ++$ID; 414 my $id = $UNIQ . ++$ID;
391 my $port = "$NODE#$id"; 415 my $port = "$NODE#$id";
392 416
393 rcv $port, shift || \&_kilme; 417 rcv $port, shift || $KILME;
394 418
395 $port 419 $port
396} 420}
397 421
398=item rcv $local_port, $callback->(@msg) 422=item rcv $local_port, $callback->(@msg)
403 427
404The global C<$SELF> (exported by this module) contains C<$port> while 428The global C<$SELF> (exported by this module) contains C<$port> while
405executing the callback. Runtime errors during callback execution will 429executing the callback. Runtime errors during callback execution will
406result in the port being C<kil>ed. 430result in the port being C<kil>ed.
407 431
408The default callback received all messages not matched by a more specific 432The default callback receives all messages not matched by a more specific
409C<tag> match. 433C<tag> match.
410 434
411=item rcv $local_port, tag => $callback->(@msg_without_tag), ... 435=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
412 436
413Register (or replace) callbacks to be called on messages starting with the 437Register (or replace) callbacks to be called on messages starting with the
668 } 692 }
669 693
670 $node->monitor ($port, $cb); 694 $node->monitor ($port, $cb);
671 695
672 defined wantarray 696 defined wantarray
673 and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }) 697 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
674} 698}
675 699
676=item $guard = mon_guard $port, $ref, $ref... 700=item $guard = mon_guard $port, $ref, $ref...
677 701
678Monitors the given C<$port> and keeps the passed references. When the port 702Monitors the given C<$port> and keeps the passed references. When the port
714will be reported as reason C<< die => $@ >>. 738will be reported as reason C<< die => $@ >>.
715 739
716Transport/communication errors are reported as C<< transport_error => 740Transport/communication errors are reported as C<< transport_error =>
717$message >>. 741$message >>.
718 742
719=cut 743Common idioms:
744
745 # silently remove yourself, do not kill linked ports
746 kil $SELF;
747
748 # report a failure in some detail
749 kil $SELF, failure_mode_1 => "it failed with too high temperature";
750
751 # do not waste much time with killing, just die when something goes wrong
752 open my $fh, "<file"
753 or die "file: $!";
720 754
721=item $port = spawn $node, $initfunc[, @initdata] 755=item $port = spawn $node, $initfunc[, @initdata]
722 756
723Creates a port on the node C<$node> (which can also be a port ID, in which 757Creates a port on the node C<$node> (which can also be a port ID, in which
724case it's the node where that port resides). 758case it's the node where that port resides).
782} 816}
783 817
784sub spawn(@) { 818sub spawn(@) {
785 my ($nodeid, undef) = split /#/, shift, 2; 819 my ($nodeid, undef) = split /#/, shift, 2;
786 820
787 my $id = "$RUNIQ." . ++$ID; 821 my $id = $RUNIQ . ++$ID;
788 822
789 $_[0] =~ /::/ 823 $_[0] =~ /::/
790 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 824 or Carp::croak "spawn init function must be a fully-qualified name, caught";
791 825
792 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_; 826 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
816 ref $action[0] 850 ref $action[0]
817 ? $action[0]() 851 ? $action[0]()
818 : snd @action; 852 : snd @action;
819 }; 853 };
820} 854}
855
856#=item $cb2 = timeout $seconds, $cb[, @args]
821 857
822=item cal $port, @msg, $callback[, $timeout] 858=item cal $port, @msg, $callback[, $timeout]
823 859
824A simple form of RPC - sends a message to the given C<$port> with the 860A simple form of RPC - sends a message to the given C<$port> with the
825given contents (C<@msg>), but adds a reply port to the message. 861given contents (C<@msg>), but adds a reply port to the message.
871 $port 907 $port
872} 908}
873 909
874=back 910=back
875 911
912=head1 DISTRIBUTED DATABASE
913
914AnyEvent::MP comes with a simple distributed database. The database will
915be mirrored asynchronously on all global nodes. Other nodes bind to one
916of the global nodes for their needs. Every node has a "local database"
917which contains all the values that are set locally. All local databases
918are merged together to form the global database, which can be queried.
919
920The database structure is that of a two-level hash - the database hash
921contains hashes which contain values, similarly to a perl hash of hashes,
922i.e.:
923
924 $DATABASE{$family}{$subkey} = $value
925
926The top level hash key is called "family", and the second-level hash key
927is called "subkey" or simply "key".
928
929The family must be alphanumeric, i.e. start with a letter and consist
930of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>,
931pretty much like Perl module names.
932
933As the family namespace is global, it is recommended to prefix family names
934with the name of the application or module using it.
935
936The subkeys must be non-empty strings, with no further restrictions.
937
938The values should preferably be strings, but other perl scalars should
939work as well (such as C<undef>, arrays and hashes).
940
941Every database entry is owned by one node - adding the same family/subkey
942combination on multiple nodes will not cause discomfort for AnyEvent::MP,
943but the result might be nondeterministic, i.e. the key might have
944different values on different nodes.
945
946Different subkeys in the same family can be owned by different nodes
947without problems, and in fact, this is the common method to create worker
948pools. For example, a worker port for image scaling might do this:
949
950 db_set my_image_scalers => $port;
951
952And clients looking for an image scaler will want to get the
953C<my_image_scalers> keys from time to time:
954
955 db_keys my_image_scalers => sub {
956 @ports = @{ $_[0] };
957 };
958
959Or better yet, they want to monitor the database family, so they always
960have a reasonable up-to-date copy:
961
962 db_mon my_image_scalers => sub {
963 @ports = keys %{ $_[0] };
964 };
965
966In general, you can set or delete single subkeys, but query and monitor
967whole families only.
968
969If you feel the need to monitor or query a single subkey, try giving it
970it's own family.
971
972=over
973
974=item $guard = db_set $family => $subkey [=> $value]
975
976Sets (or replaces) a key to the database - if C<$value> is omitted,
977C<undef> is used instead.
978
979When called in non-void context, C<db_set> returns a guard that
980automatically calls C<db_del> when it is destroyed.
981
982=item db_del $family => $subkey...
983
984Deletes one or more subkeys from the database family.
985
986=item $guard = db_reg $family => $port => $value
987
988=item $guard = db_reg $family => $port
989
990=item $guard = db_reg $family
991
992Registers a port in the given family and optionally returns a guard to
993remove it.
994
995This function basically does the same as:
996
997 db_set $family => $port => $value
998
999Except that the port is monitored and automatically removed from the
1000database family when it is kil'ed.
1001
1002If C<$value> is missing, C<undef> is used. If C<$port> is missing, then
1003C<$SELF> is used.
1004
1005This function is most useful to register a port in some port group (which
1006is just another name for a database family), and have it removed when the
1007port is gone. This works best when the port is a local port.
1008
1009=cut
1010
1011sub db_reg($$;$) {
1012 my $family = shift;
1013 my $port = @_ ? shift : $SELF;
1014
1015 my $clr = sub { db_del $family => $port };
1016 mon $port, $clr;
1017
1018 db_set $family => $port => $_[0];
1019
1020 defined wantarray
1021 and &Guard::guard ($clr)
1022}
1023
1024=item db_family $family => $cb->(\%familyhash)
1025
1026Queries the named database C<$family> and call the callback with the
1027family represented as a hash. You can keep and freely modify the hash.
1028
1029=item db_keys $family => $cb->(\@keys)
1030
1031Same as C<db_family>, except it only queries the family I<subkeys> and passes
1032them as array reference to the callback.
1033
1034=item db_values $family => $cb->(\@values)
1035
1036Same as C<db_family>, except it only queries the family I<values> and passes them
1037as array reference to the callback.
1038
1039=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
1040
1041Creates a monitor on the given database family. Each time a key is set
1042or or is deleted the callback is called with a hash containing the
1043database family and three lists of added, changed and deleted subkeys,
1044respectively. If no keys have changed then the array reference might be
1045C<undef> or even missing.
1046
1047If not called in void context, a guard object is returned that, when
1048destroyed, stops the monitor.
1049
1050The family hash reference and the key arrays belong to AnyEvent::MP and
1051B<must not be modified or stored> by the callback. When in doubt, make a
1052copy.
1053
1054As soon as possible after the monitoring starts, the callback will be
1055called with the intiial contents of the family, even if it is empty,
1056i.e. there will always be a timely call to the callback with the current
1057contents.
1058
1059It is possible that the callback is called with a change event even though
1060the subkey is already present and the value has not changed.
1061
1062The monitoring stops when the guard object is destroyed.
1063
1064Example: on every change to the family "mygroup", print out all keys.
1065
1066 my $guard = db_mon mygroup => sub {
1067 my ($family, $a, $c, $d) = @_;
1068 print "mygroup members: ", (join " ", keys %$family), "\n";
1069 };
1070
1071Exmaple: wait until the family "My::Module::workers" is non-empty.
1072
1073 my $guard; $guard = db_mon My::Module::workers => sub {
1074 my ($family, $a, $c, $d) = @_;
1075 return unless %$family;
1076 undef $guard;
1077 print "My::Module::workers now nonempty\n";
1078 };
1079
1080Example: print all changes to the family "AnyRvent::Fantasy::Module".
1081
1082 my $guard = db_mon AnyRvent::Fantasy::Module => sub {
1083 my ($family, $a, $c, $d) = @_;
1084
1085 print "+$_=$family->{$_}\n" for @$a;
1086 print "*$_=$family->{$_}\n" for @$c;
1087 print "-$_=$family->{$_}\n" for @$d;
1088 };
1089
1090=cut
1091
1092=back
1093
876=head1 AnyEvent::MP vs. Distributed Erlang 1094=head1 AnyEvent::MP vs. Distributed Erlang
877 1095
878AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 1096AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
879== aemp node, Erlang process == aemp port), so many of the documents and 1097== aemp node, Erlang process == aemp port), so many of the documents and
880programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 1098programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
1034Keeping your messages simple, concentrating on data structures rather than 1252Keeping your messages simple, concentrating on data structures rather than
1035objects, will keep your messages clean, tidy and efficient. 1253objects, will keep your messages clean, tidy and efficient.
1036 1254
1037=back 1255=back
1038 1256
1257=head1 PORTING FROM AnyEvent::MP VERSION 1.X
1258
1259AEMP version 2 has three major incompatible changes compared to version 1:
1260
1261=over 4
1262
1263=item AnyEvent::MP::Global no longer has group management functions.
1264
1265AnyEvent::MP now comes with a distributed database that is more
1266powerful. It's database families map closely to ports, but the API has
1267minor differences:
1268
1269 grp_reg $group, $port # old
1270 db_reg $group, $port # new
1271
1272 $list = grp_get $group # old
1273 db_keys $group, sub { my $list = shift } # new
1274
1275 grp_mon $group, $cb->(\@ports, $add, $del) # old
1276 db_mon $group, $cb->(\%ports, $add, $change, $del) # new
1277
1278C<grp_reg> is a no-brainer (just replace by C<db_reg>), but C<grp_get>
1279is no longer instant, because the local node might not have a copy of
1280the group. This can be partially remedied by using C<db_mon> to keep an
1281updated copy of the group:
1282
1283 my $local_group_copy;
1284 db_mon $group => sub { $local_group_copy = shift };
1285
1286 # no keys %$local_group_copy always returns the most up-to-date
1287 # list of ports in the group.
1288
1289C<grp_mon> can almost be replaced by C<db_mon>:
1290
1291 db_mon $group => sub {
1292 my ($ports, $add, $chg, $lde) = @_;
1293 $ports = [keys %$ports];
1294
1295 # now $ports, $add and $del are the same as
1296 # were originally passed by grp_mon.
1297 ...
1298 };
1299
1300=item Nodes not longer connect to all other nodes.
1301
1302In AEMP 1.x, every node automatically loads the L<AnyEvent::MP::Global>
1303module, which in turn would create connections to all other nodes in the
1304network (helped by the seed nodes).
1305
1306In version 2.x, global nodes still connect to all other global nodes, but
1307other nodes don't - now every node either is a global node itself, or
1308attaches itself to another global node.
1309
1310If a node isn't a global node itself, then it attaches itself to one
1311of its seed nodes. If that seed node isn't a global node yet, it will
1312automatically be upgraded to a global node.
1313
1314So in many cases, nothing needs to be changed - one just has to make sure
1315that all seed nodes are meshed together with the other seed nodes (as with
1316AEMP 1.x), and other nodes specify them as seed nodes.
1317
1318Not opening a connection to every other node is usually an advantage,
1319except when you need the lower latency of an already established
1320connection. To ensure a node establishes a connection to another node,
1321you can monitor the node port (C<mon $node, ...>), which will attempt to
1322create the connection (And notify you when the connection fails).
1323
1324=item Listener-less nodes are gone.
1325
1326And are not coming back, at least not in their old form.
1327
1328There are vague plans to implement some form of routing domains, which
1329might or might not bring back listener-less nodes, but don't count on it.
1330
1331The fact that most connections are now optional somewhat mitigates this,
1332as a node can be effectively unreachable from the outside without any
1333problems, as long as it isn't a global node and only reaches out to other
1334nodes (as opposed to being contacted from other nodes).
1335
1336=back
1337
1039=head1 SEE ALSO 1338=head1 SEE ALSO
1040 1339
1041L<AnyEvent::MP::Intro> - a gentle introduction. 1340L<AnyEvent::MP::Intro> - a gentle introduction.
1042 1341
1043L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 1342L<AnyEvent::MP::Kernel> - more, lower-level, stuff.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines