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.125 by root, Sat Mar 3 13:07:19 2012 UTC vs.
Revision 1.142 by root, Fri Mar 23 13:44:01 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
48 my $timer = AE::timer 1, 0, psub { 48 my $timer = AE::timer 1, 0, psub {
49 die "kill the port, delayed"; 49 die "kill the port, delayed";
50 }; 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.
59 51
60=head1 DESCRIPTION 52=head1 DESCRIPTION
61 53
62This module (-family) implements a simple message passing framework. 54This module (-family) implements a simple message passing framework.
63 55
184 176
185use common::sense; 177use common::sense;
186 178
187use Carp (); 179use Carp ();
188 180
189use AE (); 181use AnyEvent ();
190use Guard (); 182use Guard ();
191 183
192use base "Exporter"; 184use base "Exporter";
193 185
194our $VERSION = $AnyEvent::MP::Config::VERSION; 186our $VERSION = $AnyEvent::MP::Config::VERSION;
197 NODE $NODE *SELF node_of after 189 NODE $NODE *SELF node_of after
198 configure 190 configure
199 snd rcv mon mon_guard kil psub peval spawn cal 191 snd rcv mon mon_guard kil psub peval spawn cal
200 port 192 port
201 db_set db_del db_reg 193 db_set db_del db_reg
194 db_mon db_family db_keys db_values
202); 195);
203 196
204our $SELF; 197our $SELF;
205 198
206sub _self_die() { 199sub _self_die() {
230 223
231This function configures a node - it must be called exactly once (or 224This function configures a node - it must be called exactly once (or
232never) before calling other AnyEvent::MP functions. 225never) before calling other AnyEvent::MP functions.
233 226
234The key/value pairs are basically the same ones as documented for the 227The key/value pairs are basically the same ones as documented for the
235F<aemp> command line utility (sans the set/del prefix), with two additions: 228F<aemp> command line utility (sans the set/del prefix), with these additions:
236 229
237=over 4 230=over 4
238 231
239=item norc => $boolean (default false) 232=item norc => $boolean (default false)
240 233
271and the values specified directly via C<configure> have lowest priority, 264and the values specified directly via C<configure> have lowest priority,
272and can only be used to specify defaults. 265and can only be used to specify defaults.
273 266
274If the profile specifies a node ID, then this will become the node ID of 267If the profile specifies a node ID, then this will become the node ID of
275this process. If not, then the profile name will be used as node ID, with 268this process. If not, then the profile name will be used as node ID, with
276a slash (C</>) attached. 269a unique randoms tring (C</%u>) appended.
277 270
278If the node ID (or profile name) ends with a slash (C</>), then a random 271The node ID can contain some C<%> sequences that are expanded: C<%n>
279string is appended to make it unique. 272is expanded to the local nodename, C<%u> is replaced by a random
273strign to make the node unique. For example, the F<aemp> commandline
274utility uses C<aemp/%n/%u> as nodename, which might expand to
275C<aemp/cerebro/ZQDGSIkRhEZQDGSIkRhE>.
280 276
281=item step 2, bind listener sockets 277=item step 2, bind listener sockets
282 278
283The next step is to look up the binds in the profile, followed by binding 279The next step is to look up the binds in the profile, followed by binding
284aemp protocol listeners on all binds specified (it is possible and valid 280aemp protocol listeners on all binds specified (it is possible and valid
301Example: become a distributed node using the local node name as profile. 297Example: become a distributed node using the local node name as profile.
302This should be the most common form of invocation for "daemon"-type nodes. 298This should be the most common form of invocation for "daemon"-type nodes.
303 299
304 configure 300 configure
305 301
306Example: become an anonymous node. This form is often used for commandline 302Example: become a semi-anonymous node. This form is often used for
307clients. 303commandline clients.
308 304
309 configure nodeid => "anon/"; 305 configure nodeid => "myscript/%n/%u";
310 306
311Example: configure a node using a profile called seed, which is suitable 307Example: configure a node using a profile called seed, which is suitable
312for a seed node as it binds on all local addresses on a fixed port (4040, 308for a seed node as it binds on all local addresses on a fixed port (4040,
313customary for aemp). 309customary for aemp).
314 310
386 382
387=cut 383=cut
388 384
389sub rcv($@); 385sub rcv($@);
390 386
391sub _kilme { 387my $KILME = sub {
392 die "received message on port without callback"; 388 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
393} 389 kil $SELF, unhandled_message => "no callback found for message '$tag'";
390};
394 391
395sub port(;&) { 392sub port(;&) {
396 my $id = $UNIQ . ++$ID; 393 my $id = $UNIQ . ++$ID;
397 my $port = "$NODE#$id"; 394 my $port = "$NODE#$id";
398 395
399 rcv $port, shift || \&_kilme; 396 rcv $port, shift || $KILME;
400 397
401 $port 398 $port
402} 399}
403 400
404=item rcv $local_port, $callback->(@msg) 401=item rcv $local_port, $callback->(@msg)
409 406
410The global C<$SELF> (exported by this module) contains C<$port> while 407The global C<$SELF> (exported by this module) contains C<$port> while
411executing the callback. Runtime errors during callback execution will 408executing the callback. Runtime errors during callback execution will
412result in the port being C<kil>ed. 409result in the port being C<kil>ed.
413 410
414The default callback received all messages not matched by a more specific 411The default callback receives all messages not matched by a more specific
415C<tag> match. 412C<tag> match.
416 413
417=item rcv $local_port, tag => $callback->(@msg_without_tag), ... 414=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
418 415
419Register (or replace) callbacks to be called on messages starting with the 416Register (or replace) callbacks to be called on messages starting with the
583 $res 580 $res
584 } 581 }
585 } 582 }
586} 583}
587 584
585=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
586
587=item $guard = mon $port # kill $SELF when $port dies
588
588=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies 589=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
589
590=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
591
592=item $guard = mon $port # kill $SELF when $port dies
593 590
594=item $guard = mon $port, $rcvport, @msg # send a message when $port dies 591=item $guard = mon $port, $rcvport, @msg # send a message when $port dies
595 592
596Monitor the given port and do something when the port is killed or 593Monitor the given port and do something when the port is killed or
597messages to it were lost, and optionally return a guard that can be used 594messages to it were lost, and optionally return a guard that can be used
598to stop monitoring again. 595to stop monitoring again.
599 596
597The first two forms distinguish between "normal" and "abnormal" kil's:
598
599In the first form (another port given), if the C<$port> is C<kil>'ed with
600a non-empty reason, the other port (C<$rcvport>) will be kil'ed with the
601same reason. That is, on "normal" kil's nothing happens, while under all
602other conditions, the other port is killed with the same reason.
603
604The second form (kill self) is the same as the first form, except that
605C<$rvport> defaults to C<$SELF>.
606
607The remaining forms don't distinguish between "normal" and "abnormal" kil's
608- it's up to the callback or receiver to check whether the C<@reason> is
609empty and act accordingly.
610
600In the first form (callback), the callback is simply called with any 611In the third form (callback), the callback is simply called with any
601number of C<@reason> elements (no @reason means that the port was deleted 612number of C<@reason> elements (empty @reason means that the port was deleted
602"normally"). Note also that I<< the callback B<must> never die >>, so use 613"normally"). Note also that I<< the callback B<must> never die >>, so use
603C<eval> if unsure. 614C<eval> if unsure.
604 615
605In the second form (another port given), the other port (C<$rcvport>)
606will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
607"normal" kils nothing happens, while under all other conditions, the other
608port is killed with the same reason.
609
610The third form (kill self) is the same as the second form, except that
611C<$rvport> defaults to C<$SELF>.
612
613In the last form (message), a message of the form C<@msg, @reason> will be 616In the last form (message), a message of the form C<$rcvport, @msg,
614C<snd>. 617@reason> will be C<snd>.
615 618
616Monitoring-actions are one-shot: once messages are lost (and a monitoring 619Monitoring-actions are one-shot: once messages are lost (and a monitoring
617alert was raised), they are removed and will not trigger again. 620alert was raised), they are removed and will not trigger again, even if it
621turns out that the port is still alive.
618 622
619As a rule of thumb, monitoring requests should always monitor a port from 623As a rule of thumb, monitoring requests should always monitor a remote
620a local port (or callback). The reason is that kill messages might get 624port locally (using a local C<$rcvport> or a callback). The reason is that
621lost, just like any other message. Another less obvious reason is that 625kill messages might get lost, just like any other message. Another less
622even monitoring requests can get lost (for example, when the connection 626obvious reason is that even monitoring requests can get lost (for example,
623to the other node goes down permanently). When monitoring a port locally 627when the connection to the other node goes down permanently). When
624these problems do not exist. 628monitoring a port locally these problems do not exist.
625 629
626C<mon> effectively guarantees that, in the absence of hardware failures, 630C<mon> effectively guarantees that, in the absence of hardware failures,
627after starting the monitor, either all messages sent to the port will 631after starting the monitor, either all messages sent to the port will
628arrive, or the monitoring action will be invoked after possible message 632arrive, or the monitoring action will be invoked after possible message
629loss has been detected. No messages will be lost "in between" (after 633loss has been detected. No messages will be lost "in between" (after
720will be reported as reason C<< die => $@ >>. 724will be reported as reason C<< die => $@ >>.
721 725
722Transport/communication errors are reported as C<< transport_error => 726Transport/communication errors are reported as C<< transport_error =>
723$message >>. 727$message >>.
724 728
725=cut 729Common idioms:
730
731 # silently remove yourself, do not kill linked ports
732 kil $SELF;
733
734 # report a failure in some detail
735 kil $SELF, failure_mode_1 => "it failed with too high temperature";
736
737 # do not waste much time with killing, just die when something goes wrong
738 open my $fh, "<file"
739 or die "file: $!";
726 740
727=item $port = spawn $node, $initfunc[, @initdata] 741=item $port = spawn $node, $initfunc[, @initdata]
728 742
729Creates a port on the node C<$node> (which can also be a port ID, in which 743Creates a port on the node C<$node> (which can also be a port ID, in which
730case it's the node where that port resides). 744case it's the node where that port resides).
822 ref $action[0] 836 ref $action[0]
823 ? $action[0]() 837 ? $action[0]()
824 : snd @action; 838 : snd @action;
825 }; 839 };
826} 840}
841
842#=item $cb2 = timeout $seconds, $cb[, @args]
827 843
828=item cal $port, @msg, $callback[, $timeout] 844=item cal $port, @msg, $callback[, $timeout]
829 845
830A simple form of RPC - sends a message to the given C<$port> with the 846A simple form of RPC - sends a message to the given C<$port> with the
831given contents (C<@msg>), but adds a reply port to the message. 847given contents (C<@msg>), but adds a reply port to the message.
880=back 896=back
881 897
882=head1 DISTRIBUTED DATABASE 898=head1 DISTRIBUTED DATABASE
883 899
884AnyEvent::MP comes with a simple distributed database. The database will 900AnyEvent::MP comes with a simple distributed database. The database will
885be mirrored asynchronously at all global nodes. Other nodes bind to one of 901be mirrored asynchronously on all global nodes. Other nodes bind to one
886the global nodes for their needs. 902of the global nodes for their needs. Every node has a "local database"
903which contains all the values that are set locally. All local databases
904are merged together to form the global database, which can be queried.
887 905
888The database consists of a two-level hash - a hash contains a hash which 906The database structure is that of a two-level hash - the database hash
889contains values. 907contains hashes which contain values, similarly to a perl hash of hashes,
908i.e.:
909
910 $DATABASE{$family}{$subkey} = $value
890 911
891The top level hash key is called "family", and the second-level hash key 912The top level hash key is called "family", and the second-level hash key
892is simply called "key". 913is called "subkey" or simply "key".
893 914
894The family must be alphanumeric, i.e. start with a letter and consist 915The family must be alphanumeric, i.e. start with a letter and consist
895of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>, 916of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>,
896pretty much like Perl module names. 917pretty much like Perl module names.
897 918
898As the family namespace is global, it is recommended to prefix family names 919As the family namespace is global, it is recommended to prefix family names
899with the name of the application or module using it. 920with the name of the application or module using it.
900 921
901The keys must be strings, with no other limitations. 922The subkeys must be non-empty strings, with no further restrictions.
902 923
903The values should preferably be strings, but other perl scalars should 924The values should preferably be strings, but other perl scalars should
904work as well (such as undef, arrays and hashes). 925work as well (such as C<undef>, arrays and hashes).
905 926
906Every database entry is owned by one node - adding the same family/key 927Every database entry is owned by one node - adding the same family/subkey
907combination on multiple nodes will not cause discomfort for AnyEvent::MP, 928combination on multiple nodes will not cause discomfort for AnyEvent::MP,
908but the result might be nondeterministic, i.e. the key might have 929but the result might be nondeterministic, i.e. the key might have
909different values on different nodes. 930different values on different nodes.
910 931
911=item db_set $family => $key => $value 932Different subkeys in the same family can be owned by different nodes
933without problems, and in fact, this is the common method to create worker
934pools. For example, a worker port for image scaling might do this:
912 935
913Sets (or replaces) a key to the database. 936 db_set my_image_scalers => $port;
914 937
915=item db_del $family => $key 938And clients looking for an image scaler will want to get the
939C<my_image_scalers> keys from time to time:
916 940
917Deletes a key from the database. 941 db_keys my_image_scalers => sub {
942 @ports = @{ $_[0] };
943 };
918 944
945Or better yet, they want to monitor the database family, so they always
946have a reasonable up-to-date copy:
947
948 db_mon my_image_scalers => sub {
949 @ports = keys %{ $_[0] };
950 };
951
952In general, you can set or delete single subkeys, but query and monitor
953whole families only.
954
955If you feel the need to monitor or query a single subkey, try giving it
956it's own family.
957
958=over
959
919=item $guard = db_reg $family => $key [=> $value] 960=item $guard = db_set $family => $subkey [=> $value]
920 961
921Sets the key on the database and returns a guard. When the guard is 962Sets (or replaces) a key to the database - if C<$value> is omitted,
922destroyed, the key is deleted from the database. If C<$value> is missing, 963C<undef> is used instead.
923then C<undef> is used. 964
965When called in non-void context, C<db_set> returns a guard that
966automatically calls C<db_del> when it is destroyed.
967
968=item db_del $family => $subkey...
969
970Deletes one or more subkeys from the database family.
971
972=item $guard = db_reg $family => $port => $value
973
974=item $guard = db_reg $family => $port
975
976=item $guard = db_reg $family
977
978Registers a port in the given family and optionally returns a guard to
979remove it.
980
981This function basically does the same as:
982
983 db_set $family => $port => $value
984
985Except that the port is monitored and automatically removed from the
986database family when it is kil'ed.
987
988If C<$value> is missing, C<undef> is used. If C<$port> is missing, then
989C<$SELF> is used.
990
991This function is most useful to register a port in some port group (which
992is just another name for a database family), and have it removed when the
993port is gone. This works best when the port is a local port.
994
995=cut
996
997sub db_reg($$;$) {
998 my $family = shift;
999 my $port = @_ ? shift : $SELF;
1000
1001 my $clr = sub { db_del $family => $port };
1002 mon $port, $clr;
1003
1004 db_set $family => $port => $_[0];
1005
1006 defined wantarray
1007 and &Guard::guard ($clr)
1008}
1009
1010=item db_family $family => $cb->(\%familyhash)
1011
1012Queries the named database C<$family> and call the callback with the
1013family represented as a hash. You can keep and freely modify the hash.
1014
1015=item db_keys $family => $cb->(\@keys)
1016
1017Same as C<db_family>, except it only queries the family I<subkeys> and passes
1018them as array reference to the callback.
1019
1020=item db_values $family => $cb->(\@values)
1021
1022Same as C<db_family>, except it only queries the family I<values> and passes them
1023as array reference to the callback.
1024
1025=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
1026
1027Creates a monitor on the given database family. Each time a key is set
1028or or is deleted the callback is called with a hash containing the
1029database family and three lists of added, changed and deleted subkeys,
1030respectively. If no keys have changed then the array reference might be
1031C<undef> or even missing.
1032
1033If not called in void context, a guard object is returned that, when
1034destroyed, stops the monitor.
1035
1036The family hash reference and the key arrays belong to AnyEvent::MP and
1037B<must not be modified or stored> by the callback. When in doubt, make a
1038copy.
1039
1040As soon as possible after the monitoring starts, the callback will be
1041called with the intiial contents of the family, even if it is empty,
1042i.e. there will always be a timely call to the callback with the current
1043contents.
1044
1045It is possible that the callback is called with a change event even though
1046the subkey is already present and the value has not changed.
1047
1048The monitoring stops when the guard object is destroyed.
1049
1050Example: on every change to the family "mygroup", print out all keys.
1051
1052 my $guard = db_mon mygroup => sub {
1053 my ($family, $a, $c, $d) = @_;
1054 print "mygroup members: ", (join " ", keys %$family), "\n";
1055 };
1056
1057Exmaple: wait until the family "My::Module::workers" is non-empty.
1058
1059 my $guard; $guard = db_mon My::Module::workers => sub {
1060 my ($family, $a, $c, $d) = @_;
1061 return unless %$family;
1062 undef $guard;
1063 print "My::Module::workers now nonempty\n";
1064 };
1065
1066Example: print all changes to the family "AnyRvent::Fantasy::Module".
1067
1068 my $guard = db_mon AnyRvent::Fantasy::Module => sub {
1069 my ($family, $a, $c, $d) = @_;
1070
1071 print "+$_=$family->{$_}\n" for @$a;
1072 print "*$_=$family->{$_}\n" for @$c;
1073 print "-$_=$family->{$_}\n" for @$d;
1074 };
924 1075
925=cut 1076=cut
926 1077
927=back 1078=back
928 1079
1087Keeping your messages simple, concentrating on data structures rather than 1238Keeping your messages simple, concentrating on data structures rather than
1088objects, will keep your messages clean, tidy and efficient. 1239objects, will keep your messages clean, tidy and efficient.
1089 1240
1090=back 1241=back
1091 1242
1243=head1 PORTING FROM AnyEvent::MP VERSION 1.X
1244
1245AEMP version 2 has a few major incompatible changes compared to version 1:
1246
1247=over 4
1248
1249=item AnyEvent::MP::Global no longer has group management functions.
1250
1251At least not officially - the grp_* functions are still exported and might
1252work, but they will be removed in some later release.
1253
1254AnyEvent::MP now comes with a distributed database that is more
1255powerful. Its database families map closely to port groups, but the API
1256has changed (the functions are also now exported by AnyEvent::MP). Here is
1257a rough porting guide:
1258
1259 grp_reg $group, $port # old
1260 db_reg $group, $port # new
1261
1262 $list = grp_get $group # old
1263 db_keys $group, sub { my $list = shift } # new
1264
1265 grp_mon $group, $cb->(\@ports, $add, $del) # old
1266 db_mon $group, $cb->(\%ports, $add, $change, $del) # new
1267
1268C<grp_reg> is a no-brainer (just replace by C<db_reg>), but C<grp_get> is
1269no longer instant, because the local node might not have a copy of the
1270group. You can either modify your code to allow for a callback, or use
1271C<db_mon> to keep an updated copy of the group:
1272
1273 my $local_group_copy;
1274 db_mon $group => sub { $local_group_copy = $_[0] };
1275
1276 # now "keys %$local_group_copy" always returns the most up-to-date
1277 # list of ports in the group.
1278
1279C<grp_mon> can be replaced by C<db_mon> with minor changes - C<db_mon>
1280passes a hash as first argument, and an extra C<$chg> argument that can be
1281ignored:
1282
1283 db_mon $group => sub {
1284 my ($ports, $add, $chg, $lde) = @_;
1285 $ports = [keys %$ports];
1286
1287 # now $ports, $add and $del are the same as
1288 # were originally passed by grp_mon.
1289 ...
1290 };
1291
1292=item Nodes not longer connect to all other nodes.
1293
1294In AEMP 1.x, every node automatically loads the L<AnyEvent::MP::Global>
1295module, which in turn would create connections to all other nodes in the
1296network (helped by the seed nodes).
1297
1298In version 2.x, global nodes still connect to all other global nodes, but
1299other nodes don't - now every node either is a global node itself, or
1300attaches itself to another global node.
1301
1302If a node isn't a global node itself, then it attaches itself to one
1303of its seed nodes. If that seed node isn't a global node yet, it will
1304automatically be upgraded to a global node.
1305
1306So in many cases, nothing needs to be changed - one just has to make sure
1307that all seed nodes are meshed together with the other seed nodes (as with
1308AEMP 1.x), and other nodes specify them as seed nodes. This is most easily
1309achieved by specifying the same set of seed nodes for all nodes in the
1310network.
1311
1312Not opening a connection to every other node is usually an advantage,
1313except when you need the lower latency of an already established
1314connection. To ensure a node establishes a connection to another node,
1315you can monitor the node port (C<mon $node, ...>), which will attempt to
1316create the connection (and notify you when the connection fails).
1317
1318=item Listener-less nodes (nodes without binds) are gone.
1319
1320And are not coming back, at least not in their old form. If no C<binds>
1321are specified for a node, AnyEvent::MP assumes a default of C<*:*>.
1322
1323There are vague plans to implement some form of routing domains, which
1324might or might not bring back listener-less nodes, but don't count on it.
1325
1326The fact that most connections are now optional somewhat mitigates this,
1327as a node can be effectively unreachable from the outside without any
1328problems, as long as it isn't a global node and only reaches out to other
1329nodes (as opposed to being contacted from other nodes).
1330
1331=item $AnyEvent::MP::Kernel::WARN has gone.
1332
1333AnyEvent has acquired a logging framework (L<AnyEvent::Log>), and AEMP now
1334uses this, and so should your programs.
1335
1336Every module now documents what kinds of messages it generates, with
1337AnyEvent::MP acting as a catch all.
1338
1339On the positive side, this means that instead of setting
1340C<PERL_ANYEVENT_MP_WARNLEVEL>, you can get away by setting C<AE_VERBOSE> -
1341much less to type.
1342
1343=back
1344
1345=head1 LOGGING
1346
1347AnyEvent::MP does not normally log anything by itself, but sinc eit is the
1348root of the contetx hierarchy for AnyEvent::MP modules, it will receive
1349all log messages by submodules.
1350
1092=head1 SEE ALSO 1351=head1 SEE ALSO
1093 1352
1094L<AnyEvent::MP::Intro> - a gentle introduction. 1353L<AnyEvent::MP::Intro> - a gentle introduction.
1095 1354
1096L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 1355L<AnyEvent::MP::Kernel> - more, lower-level, stuff.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines