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.123 by root, Thu Mar 1 19:37:59 2012 UTC vs.
Revision 1.139 by root, Thu Mar 22 20:07:31 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
185use common::sense; 185use common::sense;
186 186
187use Carp (); 187use Carp ();
188 188
189use AE (); 189use AE ();
190use Guard ();
190 191
191use base "Exporter"; 192use base "Exporter";
192 193
193our $VERSION = $AnyEvent::MP::Config::VERSION; 194our $VERSION = $AnyEvent::MP::Config::VERSION;
194 195
195our @EXPORT = qw( 196our @EXPORT = qw(
196 NODE $NODE *SELF node_of after 197 NODE $NODE *SELF node_of after
197 configure 198 configure
198 snd rcv mon mon_guard kil psub peval spawn cal 199 snd rcv mon mon_guard kil psub peval spawn cal
199 port 200 port
201 db_set db_del db_reg
202 db_mon db_family db_keys db_values
200); 203);
201 204
202our $SELF; 205our $SELF;
203 206
204sub _self_die() { 207sub _self_die() {
228 231
229This function configures a node - it must be called exactly once (or 232This function configures a node - it must be called exactly once (or
230never) before calling other AnyEvent::MP functions. 233never) before calling other AnyEvent::MP functions.
231 234
232The 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
233F<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:
234 237
235=over 4 238=over 4
236 239
237=item norc => $boolean (default false) 240=item norc => $boolean (default false)
238 241
243=item force => $boolean (default false) 246=item force => $boolean (default false)
244 247
245IF true, then the values specified in the C<configure> will take 248IF true, then the values specified in the C<configure> will take
246precedence 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
247the 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.
248 264
249=back 265=back
250 266
251=over 4 267=over 4
252 268
269and the values specified directly via C<configure> have lowest priority, 285and the values specified directly via C<configure> have lowest priority,
270and can only be used to specify defaults. 286and can only be used to specify defaults.
271 287
272If 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
273this process. If not, then the profile name will be used as node ID, with 289this process. If not, then the profile name will be used as node ID, with
274a slash (C</>) attached. 290a unique randoms tring (C</%u>) appended.
275 291
276If the node ID (or profile name) ends with a slash (C</>), then a random 292The node ID can contain some C<%> sequences that are expanded: C<%n>
277string is appended to make it unique. 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>.
278 297
279=item step 2, bind listener sockets 298=item step 2, bind listener sockets
280 299
281The 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
282aemp protocol listeners on all binds specified (it is possible and valid 301aemp protocol listeners on all binds specified (it is possible and valid
299Example: become a distributed node using the local node name as profile. 318Example: become a distributed node using the local node name as profile.
300This 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.
301 320
302 configure 321 configure
303 322
304Example: become an anonymous node. This form is often used for commandline 323Example: become a semi-anonymous node. This form is often used for
305clients. 324commandline clients.
306 325
307 configure nodeid => "anon/"; 326 configure nodeid => "myscript/%n/%u";
308 327
309Example: configure a node using a profile called seed, which is suitable 328Example: configure a node using a profile called seed, which is suitable
310for 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,
311customary for aemp). 330customary for aemp).
312 331
384 403
385=cut 404=cut
386 405
387sub rcv($@); 406sub rcv($@);
388 407
389sub _kilme { 408my $KILME = sub {
390 die "received message on port without callback"; 409 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
391} 410 kil $SELF, unhandled_message => "no callback found for message '$tag'";
411};
392 412
393sub port(;&) { 413sub port(;&) {
394 my $id = $UNIQ . ++$ID; 414 my $id = $UNIQ . ++$ID;
395 my $port = "$NODE#$id"; 415 my $port = "$NODE#$id";
396 416
397 rcv $port, shift || \&_kilme; 417 rcv $port, shift || $KILME;
398 418
399 $port 419 $port
400} 420}
401 421
402=item rcv $local_port, $callback->(@msg) 422=item rcv $local_port, $callback->(@msg)
407 427
408The global C<$SELF> (exported by this module) contains C<$port> while 428The global C<$SELF> (exported by this module) contains C<$port> while
409executing the callback. Runtime errors during callback execution will 429executing the callback. Runtime errors during callback execution will
410result in the port being C<kil>ed. 430result in the port being C<kil>ed.
411 431
412The default callback received all messages not matched by a more specific 432The default callback receives all messages not matched by a more specific
413C<tag> match. 433C<tag> match.
414 434
415=item rcv $local_port, tag => $callback->(@msg_without_tag), ... 435=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
416 436
417Register (or replace) callbacks to be called on messages starting with the 437Register (or replace) callbacks to be called on messages starting with the
581 $res 601 $res
582 } 602 }
583 } 603 }
584} 604}
585 605
606=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
607
608=item $guard = mon $port # kill $SELF when $port dies
609
586=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies 610=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
587
588=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
589
590=item $guard = mon $port # kill $SELF when $port dies
591 611
592=item $guard = mon $port, $rcvport, @msg # send a message when $port dies 612=item $guard = mon $port, $rcvport, @msg # send a message when $port dies
593 613
594Monitor the given port and do something when the port is killed or 614Monitor the given port and do something when the port is killed or
595messages to it were lost, and optionally return a guard that can be used 615messages to it were lost, and optionally return a guard that can be used
596to stop monitoring again. 616to stop monitoring again.
597 617
618The first two forms distinguish between "normal" and "abnormal" kil's:
619
620In the first form (another port given), if the C<$port> is C<kil>'ed with
621a non-empty reason, the other port (C<$rcvport>) will be kil'ed with the
622same reason. That is, on "normal" kil's nothing happens, while under all
623other conditions, the other port is killed with the same reason.
624
625The second form (kill self) is the same as the first form, except that
626C<$rvport> defaults to C<$SELF>.
627
628The remaining forms don't distinguish between "normal" and "abnormal" kil's
629- it's up to the callback or receiver to check whether the C<@reason> is
630empty and act accordingly.
631
598In the first form (callback), the callback is simply called with any 632In the third form (callback), the callback is simply called with any
599number of C<@reason> elements (no @reason means that the port was deleted 633number of C<@reason> elements (empty @reason means that the port was deleted
600"normally"). Note also that I<< the callback B<must> never die >>, so use 634"normally"). Note also that I<< the callback B<must> never die >>, so use
601C<eval> if unsure. 635C<eval> if unsure.
602 636
603In the second form (another port given), the other port (C<$rcvport>)
604will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
605"normal" kils nothing happens, while under all other conditions, the other
606port is killed with the same reason.
607
608The third form (kill self) is the same as the second form, except that
609C<$rvport> defaults to C<$SELF>.
610
611In the last form (message), a message of the form C<@msg, @reason> will be 637In the last form (message), a message of the form C<$rcvport, @msg,
612C<snd>. 638@reason> will be C<snd>.
613 639
614Monitoring-actions are one-shot: once messages are lost (and a monitoring 640Monitoring-actions are one-shot: once messages are lost (and a monitoring
615alert was raised), they are removed and will not trigger again. 641alert was raised), they are removed and will not trigger again, even if it
642turns out that the port is still alive.
616 643
617As a rule of thumb, monitoring requests should always monitor a port from 644As a rule of thumb, monitoring requests should always monitor a remote
618a local port (or callback). The reason is that kill messages might get 645port locally (using a local C<$rcvport> or a callback). The reason is that
619lost, just like any other message. Another less obvious reason is that 646kill messages might get lost, just like any other message. Another less
620even monitoring requests can get lost (for example, when the connection 647obvious reason is that even monitoring requests can get lost (for example,
621to the other node goes down permanently). When monitoring a port locally 648when the connection to the other node goes down permanently). When
622these problems do not exist. 649monitoring a port locally these problems do not exist.
623 650
624C<mon> effectively guarantees that, in the absence of hardware failures, 651C<mon> effectively guarantees that, in the absence of hardware failures,
625after starting the monitor, either all messages sent to the port will 652after starting the monitor, either all messages sent to the port will
626arrive, or the monitoring action will be invoked after possible message 653arrive, or the monitoring action will be invoked after possible message
627loss has been detected. No messages will be lost "in between" (after 654loss has been detected. No messages will be lost "in between" (after
672 } 699 }
673 700
674 $node->monitor ($port, $cb); 701 $node->monitor ($port, $cb);
675 702
676 defined wantarray 703 defined wantarray
677 and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }) 704 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
678} 705}
679 706
680=item $guard = mon_guard $port, $ref, $ref... 707=item $guard = mon_guard $port, $ref, $ref...
681 708
682Monitors the given C<$port> and keeps the passed references. When the port 709Monitors the given C<$port> and keeps the passed references. When the port
718will be reported as reason C<< die => $@ >>. 745will be reported as reason C<< die => $@ >>.
719 746
720Transport/communication errors are reported as C<< transport_error => 747Transport/communication errors are reported as C<< transport_error =>
721$message >>. 748$message >>.
722 749
723=cut 750Common idioms:
751
752 # silently remove yourself, do not kill linked ports
753 kil $SELF;
754
755 # report a failure in some detail
756 kil $SELF, failure_mode_1 => "it failed with too high temperature";
757
758 # do not waste much time with killing, just die when something goes wrong
759 open my $fh, "<file"
760 or die "file: $!";
724 761
725=item $port = spawn $node, $initfunc[, @initdata] 762=item $port = spawn $node, $initfunc[, @initdata]
726 763
727Creates a port on the node C<$node> (which can also be a port ID, in which 764Creates a port on the node C<$node> (which can also be a port ID, in which
728case it's the node where that port resides). 765case it's the node where that port resides).
820 ref $action[0] 857 ref $action[0]
821 ? $action[0]() 858 ? $action[0]()
822 : snd @action; 859 : snd @action;
823 }; 860 };
824} 861}
862
863#=item $cb2 = timeout $seconds, $cb[, @args]
825 864
826=item cal $port, @msg, $callback[, $timeout] 865=item cal $port, @msg, $callback[, $timeout]
827 866
828A simple form of RPC - sends a message to the given C<$port> with the 867A simple form of RPC - sends a message to the given C<$port> with the
829given contents (C<@msg>), but adds a reply port to the message. 868given contents (C<@msg>), but adds a reply port to the message.
875 $port 914 $port
876} 915}
877 916
878=back 917=back
879 918
919=head1 DISTRIBUTED DATABASE
920
921AnyEvent::MP comes with a simple distributed database. The database will
922be mirrored asynchronously on all global nodes. Other nodes bind to one
923of the global nodes for their needs. Every node has a "local database"
924which contains all the values that are set locally. All local databases
925are merged together to form the global database, which can be queried.
926
927The database structure is that of a two-level hash - the database hash
928contains hashes which contain values, similarly to a perl hash of hashes,
929i.e.:
930
931 $DATABASE{$family}{$subkey} = $value
932
933The top level hash key is called "family", and the second-level hash key
934is called "subkey" or simply "key".
935
936The family must be alphanumeric, i.e. start with a letter and consist
937of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>,
938pretty much like Perl module names.
939
940As the family namespace is global, it is recommended to prefix family names
941with the name of the application or module using it.
942
943The subkeys must be non-empty strings, with no further restrictions.
944
945The values should preferably be strings, but other perl scalars should
946work as well (such as C<undef>, arrays and hashes).
947
948Every database entry is owned by one node - adding the same family/subkey
949combination on multiple nodes will not cause discomfort for AnyEvent::MP,
950but the result might be nondeterministic, i.e. the key might have
951different values on different nodes.
952
953Different subkeys in the same family can be owned by different nodes
954without problems, and in fact, this is the common method to create worker
955pools. For example, a worker port for image scaling might do this:
956
957 db_set my_image_scalers => $port;
958
959And clients looking for an image scaler will want to get the
960C<my_image_scalers> keys from time to time:
961
962 db_keys my_image_scalers => sub {
963 @ports = @{ $_[0] };
964 };
965
966Or better yet, they want to monitor the database family, so they always
967have a reasonable up-to-date copy:
968
969 db_mon my_image_scalers => sub {
970 @ports = keys %{ $_[0] };
971 };
972
973In general, you can set or delete single subkeys, but query and monitor
974whole families only.
975
976If you feel the need to monitor or query a single subkey, try giving it
977it's own family.
978
979=over
980
981=item $guard = db_set $family => $subkey [=> $value]
982
983Sets (or replaces) a key to the database - if C<$value> is omitted,
984C<undef> is used instead.
985
986When called in non-void context, C<db_set> returns a guard that
987automatically calls C<db_del> when it is destroyed.
988
989=item db_del $family => $subkey...
990
991Deletes one or more subkeys from the database family.
992
993=item $guard = db_reg $family => $port => $value
994
995=item $guard = db_reg $family => $port
996
997=item $guard = db_reg $family
998
999Registers a port in the given family and optionally returns a guard to
1000remove it.
1001
1002This function basically does the same as:
1003
1004 db_set $family => $port => $value
1005
1006Except that the port is monitored and automatically removed from the
1007database family when it is kil'ed.
1008
1009If C<$value> is missing, C<undef> is used. If C<$port> is missing, then
1010C<$SELF> is used.
1011
1012This function is most useful to register a port in some port group (which
1013is just another name for a database family), and have it removed when the
1014port is gone. This works best when the port is a local port.
1015
1016=cut
1017
1018sub db_reg($$;$) {
1019 my $family = shift;
1020 my $port = @_ ? shift : $SELF;
1021
1022 my $clr = sub { db_del $family => $port };
1023 mon $port, $clr;
1024
1025 db_set $family => $port => $_[0];
1026
1027 defined wantarray
1028 and &Guard::guard ($clr)
1029}
1030
1031=item db_family $family => $cb->(\%familyhash)
1032
1033Queries the named database C<$family> and call the callback with the
1034family represented as a hash. You can keep and freely modify the hash.
1035
1036=item db_keys $family => $cb->(\@keys)
1037
1038Same as C<db_family>, except it only queries the family I<subkeys> and passes
1039them as array reference to the callback.
1040
1041=item db_values $family => $cb->(\@values)
1042
1043Same as C<db_family>, except it only queries the family I<values> and passes them
1044as array reference to the callback.
1045
1046=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
1047
1048Creates a monitor on the given database family. Each time a key is set
1049or or is deleted the callback is called with a hash containing the
1050database family and three lists of added, changed and deleted subkeys,
1051respectively. If no keys have changed then the array reference might be
1052C<undef> or even missing.
1053
1054If not called in void context, a guard object is returned that, when
1055destroyed, stops the monitor.
1056
1057The family hash reference and the key arrays belong to AnyEvent::MP and
1058B<must not be modified or stored> by the callback. When in doubt, make a
1059copy.
1060
1061As soon as possible after the monitoring starts, the callback will be
1062called with the intiial contents of the family, even if it is empty,
1063i.e. there will always be a timely call to the callback with the current
1064contents.
1065
1066It is possible that the callback is called with a change event even though
1067the subkey is already present and the value has not changed.
1068
1069The monitoring stops when the guard object is destroyed.
1070
1071Example: on every change to the family "mygroup", print out all keys.
1072
1073 my $guard = db_mon mygroup => sub {
1074 my ($family, $a, $c, $d) = @_;
1075 print "mygroup members: ", (join " ", keys %$family), "\n";
1076 };
1077
1078Exmaple: wait until the family "My::Module::workers" is non-empty.
1079
1080 my $guard; $guard = db_mon My::Module::workers => sub {
1081 my ($family, $a, $c, $d) = @_;
1082 return unless %$family;
1083 undef $guard;
1084 print "My::Module::workers now nonempty\n";
1085 };
1086
1087Example: print all changes to the family "AnyRvent::Fantasy::Module".
1088
1089 my $guard = db_mon AnyRvent::Fantasy::Module => sub {
1090 my ($family, $a, $c, $d) = @_;
1091
1092 print "+$_=$family->{$_}\n" for @$a;
1093 print "*$_=$family->{$_}\n" for @$c;
1094 print "-$_=$family->{$_}\n" for @$d;
1095 };
1096
1097=cut
1098
1099=back
1100
880=head1 AnyEvent::MP vs. Distributed Erlang 1101=head1 AnyEvent::MP vs. Distributed Erlang
881 1102
882AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 1103AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
883== aemp node, Erlang process == aemp port), so many of the documents and 1104== aemp node, Erlang process == aemp port), so many of the documents and
884programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 1105programming techniques employed by Erlang apply to AnyEvent::MP. Here is a
1038Keeping your messages simple, concentrating on data structures rather than 1259Keeping your messages simple, concentrating on data structures rather than
1039objects, will keep your messages clean, tidy and efficient. 1260objects, will keep your messages clean, tidy and efficient.
1040 1261
1041=back 1262=back
1042 1263
1264=head1 PORTING FROM AnyEvent::MP VERSION 1.X
1265
1266AEMP version 2 has a few major incompatible changes compared to version 1:
1267
1268=over 4
1269
1270=item AnyEvent::MP::Global no longer has group management functions.
1271
1272AnyEvent::MP now comes with a distributed database that is more
1273powerful. Its database families map closely to port groups, but the API
1274has changed (the functions are also now exported by AnyEvent::MP). Here is
1275a rough porting guide:
1276
1277 grp_reg $group, $port # old
1278 db_reg $group, $port # new
1279
1280 $list = grp_get $group # old
1281 db_keys $group, sub { my $list = shift } # new
1282
1283 grp_mon $group, $cb->(\@ports, $add, $del) # old
1284 db_mon $group, $cb->(\%ports, $add, $change, $del) # new
1285
1286C<grp_reg> is a no-brainer (just replace by C<db_reg>), but C<grp_get> is
1287no longer instant, because the local node might not have a copy of the
1288group. You can either modify your code to allow for a callback, or use
1289C<db_mon> to keep an updated copy of the group:
1290
1291 my $local_group_copy;
1292 db_mon $group => sub { $local_group_copy = $_[0] };
1293
1294 # now "keys %$local_group_copy" always returns the most up-to-date
1295 # list of ports in the group.
1296
1297C<grp_mon> can be replaced by C<db_mon> with minor changes - C<db_mon>
1298passes a hash as first argument, and an extra C<$chg> argument that can be
1299ignored:
1300
1301 db_mon $group => sub {
1302 my ($ports, $add, $chg, $lde) = @_;
1303 $ports = [keys %$ports];
1304
1305 # now $ports, $add and $del are the same as
1306 # were originally passed by grp_mon.
1307 ...
1308 };
1309
1310=item Nodes not longer connect to all other nodes.
1311
1312In AEMP 1.x, every node automatically loads the L<AnyEvent::MP::Global>
1313module, which in turn would create connections to all other nodes in the
1314network (helped by the seed nodes).
1315
1316In version 2.x, global nodes still connect to all other global nodes, but
1317other nodes don't - now every node either is a global node itself, or
1318attaches itself to another global node.
1319
1320If a node isn't a global node itself, then it attaches itself to one
1321of its seed nodes. If that seed node isn't a global node yet, it will
1322automatically be upgraded to a global node.
1323
1324So in many cases, nothing needs to be changed - one just has to make sure
1325that all seed nodes are meshed together with the other seed nodes (as with
1326AEMP 1.x), and other nodes specify them as seed nodes. This is most easily
1327achieved by specifying the same set of seed nodes for all nodes in the
1328network.
1329
1330Not opening a connection to every other node is usually an advantage,
1331except when you need the lower latency of an already established
1332connection. To ensure a node establishes a connection to another node,
1333you can monitor the node port (C<mon $node, ...>), which will attempt to
1334create the connection (and notify you when the connection fails).
1335
1336=item Listener-less nodes (nodes without binds) are gone.
1337
1338And are not coming back, at least not in their old form. If no C<binds>
1339are specified for a node, AnyEvent::MP assumes a default of C<*:*>.
1340
1341There are vague plans to implement some form of routing domains, which
1342might or might not bring back listener-less nodes, but don't count on it.
1343
1344The fact that most connections are now optional somewhat mitigates this,
1345as a node can be effectively unreachable from the outside without any
1346problems, as long as it isn't a global node and only reaches out to other
1347nodes (as opposed to being contacted from other nodes).
1348
1349=item $AnyEvent::MP::Kernel::WARN has gone.
1350
1351AnyEvent has acquired a logging framework (L<AnyEvent::Log>), and AEMP now
1352uses this, and so should your programs.
1353
1354Every module now documents what kinds of messages it generates, with
1355AnyEvent::MP acting as a catch all.
1356
1357On the positive side, this means that instead of setting
1358C<PERL_ANYEVENT_MP_WARNLEVEL>, you can get away by setting C<AE_VERBOSE> -
1359much less to type.
1360
1361=back
1362
1363=head1 LOGGING
1364
1365AnyEvent::MP does not normally log anything by itself, but sinc eit is the
1366root of the contetx hierarchy for AnyEvent::MP modules, it will receive
1367all log messages by submodules.
1368
1043=head1 SEE ALSO 1369=head1 SEE ALSO
1044 1370
1045L<AnyEvent::MP::Intro> - a gentle introduction. 1371L<AnyEvent::MP::Intro> - a gentle introduction.
1046 1372
1047L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 1373L<AnyEvent::MP::Kernel> - more, lower-level, stuff.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines