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.122 by root, Wed Feb 29 18:44:59 2012 UTC vs.
Revision 1.133 by root, Mon Mar 12 10:34:06 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->($nodeid)
253
254In addition to specifying a boolean, you can specify a code reference that
255is called for every remote execution attempt - the execution request is
256granted iff the callback returns a true value.
257
258See F<semp setsecure> for more info.
247 259
248=back 260=back
249 261
250=over 4 262=over 4
251 263
268and the values specified directly via C<configure> have lowest priority, 280and the values specified directly via C<configure> have lowest priority,
269and can only be used to specify defaults. 281and can only be used to specify defaults.
270 282
271If the profile specifies a node ID, then this will become the node ID of 283If 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 284this process. If not, then the profile name will be used as node ID, with
273a slash (C</>) attached. 285a unique randoms tring (C</%u>) appended.
274 286
275If the node ID (or profile name) ends with a slash (C</>), then a random 287The node ID can contain some C<%> sequences that are expanded: C<%n>
276string is appended to make it unique. 288is expanded to the local nodename, C<%u> is replaced by a random
289strign to make the node unique. For example, the F<aemp> commandline
290utility uses C<aemp/%n/%u> as nodename, which might expand to
291C<aemp/cerebro/ZQDGSIkRhEZQDGSIkRhE>.
277 292
278=item step 2, bind listener sockets 293=item step 2, bind listener sockets
279 294
280The next step is to look up the binds in the profile, followed by binding 295The 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 296aemp protocol listeners on all binds specified (it is possible and valid
298Example: become a distributed node using the local node name as profile. 313Example: become a distributed node using the local node name as profile.
299This should be the most common form of invocation for "daemon"-type nodes. 314This should be the most common form of invocation for "daemon"-type nodes.
300 315
301 configure 316 configure
302 317
303Example: become an anonymous node. This form is often used for commandline 318Example: become a semi-anonymous node. This form is often used for
304clients. 319commandline clients.
305 320
306 configure nodeid => "anon/"; 321 configure nodeid => "myscript/%n/%u";
307 322
308Example: configure a node using a profile called seed, which is suitable 323Example: 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, 324for a seed node as it binds on all local addresses on a fixed port (4040,
310customary for aemp). 325customary for aemp).
311 326
383 398
384=cut 399=cut
385 400
386sub rcv($@); 401sub rcv($@);
387 402
388sub _kilme { 403my $KILME = sub {
389 die "received message on port without callback"; 404 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
390} 405 kil $SELF, unhandled_message => "no callback set for message (first element $tag)";
406};
391 407
392sub port(;&) { 408sub port(;&) {
393 my $id = "$UNIQ." . ++$ID; 409 my $id = $UNIQ . ++$ID;
394 my $port = "$NODE#$id"; 410 my $port = "$NODE#$id";
395 411
396 rcv $port, shift || \&_kilme; 412 rcv $port, shift || $KILME;
397 413
398 $port 414 $port
399} 415}
400 416
401=item rcv $local_port, $callback->(@msg) 417=item rcv $local_port, $callback->(@msg)
406 422
407The global C<$SELF> (exported by this module) contains C<$port> while 423The global C<$SELF> (exported by this module) contains C<$port> while
408executing the callback. Runtime errors during callback execution will 424executing the callback. Runtime errors during callback execution will
409result in the port being C<kil>ed. 425result in the port being C<kil>ed.
410 426
411The default callback received all messages not matched by a more specific 427The default callback receives all messages not matched by a more specific
412C<tag> match. 428C<tag> match.
413 429
414=item rcv $local_port, tag => $callback->(@msg_without_tag), ... 430=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
415 431
416Register (or replace) callbacks to be called on messages starting with the 432Register (or replace) callbacks to be called on messages starting with the
671 } 687 }
672 688
673 $node->monitor ($port, $cb); 689 $node->monitor ($port, $cb);
674 690
675 defined wantarray 691 defined wantarray
676 and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }) 692 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
677} 693}
678 694
679=item $guard = mon_guard $port, $ref, $ref... 695=item $guard = mon_guard $port, $ref, $ref...
680 696
681Monitors the given C<$port> and keeps the passed references. When the port 697Monitors the given C<$port> and keeps the passed references. When the port
717will be reported as reason C<< die => $@ >>. 733will be reported as reason C<< die => $@ >>.
718 734
719Transport/communication errors are reported as C<< transport_error => 735Transport/communication errors are reported as C<< transport_error =>
720$message >>. 736$message >>.
721 737
722=cut 738Common idioms:
739
740 # silently remove yourself, do not kill linked ports
741 kil $SELF;
742
743 # report a failure in some detail
744 kil $SELF, failure_mode_1 => "it failed with too high temperature";
745
746 # do not waste much time with killing, just die when something goes wrong
747 open my $fh, "<file"
748 or die "file: $!";
723 749
724=item $port = spawn $node, $initfunc[, @initdata] 750=item $port = spawn $node, $initfunc[, @initdata]
725 751
726Creates a port on the node C<$node> (which can also be a port ID, in which 752Creates 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). 753case it's the node where that port resides).
785} 811}
786 812
787sub spawn(@) { 813sub spawn(@) {
788 my ($nodeid, undef) = split /#/, shift, 2; 814 my ($nodeid, undef) = split /#/, shift, 2;
789 815
790 my $id = "$RUNIQ." . ++$ID; 816 my $id = $RUNIQ . ++$ID;
791 817
792 $_[0] =~ /::/ 818 $_[0] =~ /::/
793 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 819 or Carp::croak "spawn init function must be a fully-qualified name, caught";
794 820
795 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_; 821 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
819 ref $action[0] 845 ref $action[0]
820 ? $action[0]() 846 ? $action[0]()
821 : snd @action; 847 : snd @action;
822 }; 848 };
823} 849}
850
851#=item $cb2 = timeout $seconds, $cb[, @args]
824 852
825=item cal $port, @msg, $callback[, $timeout] 853=item cal $port, @msg, $callback[, $timeout]
826 854
827A simple form of RPC - sends a message to the given C<$port> with the 855A 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. 856given contents (C<@msg>), but adds a reply port to the message.
874 $port 902 $port
875} 903}
876 904
877=back 905=back
878 906
907=head1 DISTRIBUTED DATABASE
908
909AnyEvent::MP comes with a simple distributed database. The database will
910be mirrored asynchronously on all global nodes. Other nodes bind to one
911of the global nodes for their needs. Every node has a "local database"
912which contains all the values that are set locally. All local databases
913are merged together to form the global database, which can be queried.
914
915The database structure is that of a two-level hash - the database hash
916contains hashes which contain values, similarly to a perl hash of hashes,
917i.e.:
918
919 $DATABASE{$family}{$subkey} = $value
920
921The top level hash key is called "family", and the second-level hash key
922is called "subkey" or simply "key".
923
924The family must be alphanumeric, i.e. start with a letter and consist
925of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>,
926pretty much like Perl module names.
927
928As the family namespace is global, it is recommended to prefix family names
929with the name of the application or module using it.
930
931The subkeys must be non-empty strings, with no further restrictions.
932
933The values should preferably be strings, but other perl scalars should
934work as well (such as C<undef>, arrays and hashes).
935
936Every database entry is owned by one node - adding the same family/subkey
937combination on multiple nodes will not cause discomfort for AnyEvent::MP,
938but the result might be nondeterministic, i.e. the key might have
939different values on different nodes.
940
941Different subkeys in the same family can be owned by different nodes
942without problems, and in fact, this is the common method to create worker
943pools. For example, a worker port for image scaling might do this:
944
945 db_set my_image_scalers => $port;
946
947And clients looking for an image scaler will want to get the
948C<my_image_scalers> keys from time to time:
949
950 db_keys my_image_scalers => sub {
951 @ports = @{ $_[0] };
952 };
953
954Or better yet, they want to monitor the database family, so they always
955have a reasonable up-to-date copy:
956
957 db_mon my_image_scalers => sub {
958 @ports = keys %{ $_[0] };
959 };
960
961In general, you can set or delete single subkeys, but query and monitor
962whole families only.
963
964If you feel the need to monitor or query a single subkey, try giving it
965it's own family.
966
967=over
968
969=item db_set $family => $subkey [=> $value]
970
971Sets (or replaces) a key to the database - if C<$value> is omitted,
972C<undef> is used instead.
973
974=item db_del $family => $subkey...
975
976Deletes one or more subkeys from the database family.
977
978=item $guard = db_reg $family => $subkey [=> $value]
979
980Sets the key on the database and returns a guard. When the guard is
981destroyed, the key is deleted from the database. If C<$value> is missing,
982then C<undef> is used.
983
984=item db_family $family => $cb->(\%familyhash)
985
986Queries the named database C<$family> and call the callback with the
987family represented as a hash. You can keep and freely modify the hash.
988
989=item db_keys $family => $cb->(\@keys)
990
991Same as C<db_family>, except it only queries the family I<subkeys> and passes
992them as array reference to the callback.
993
994=item db_values $family => $cb->(\@values)
995
996Same as C<db_family>, except it only queries the family I<values> and passes them
997as array reference to the callback.
998
999=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
1000
1001Creates a monitor on the given database family. Each time a key is set
1002or or is deleted the callback is called with a hash containing the
1003database family and three lists of added, changed and deleted subkeys,
1004respectively. If no keys have changed then the array reference might be
1005C<undef> or even missing.
1006
1007If not called in void context, a guard object is returned that, when
1008destroyed, stops the monitor.
1009
1010The family hash reference and the key arrays belong to AnyEvent::MP and
1011B<must not be modified or stored> by the callback. When in doubt, make a
1012copy.
1013
1014As soon as possible after the monitoring starts, the callback will be
1015called with the intiial contents of the family, even if it is empty,
1016i.e. there will always be a timely call to the callback with the current
1017contents.
1018
1019It is possible that the callback is called with a change event even though
1020the subkey is already present and the value has not changed.
1021
1022The monitoring stops when the guard object is destroyed.
1023
1024Example: on every change to the family "mygroup", print out all keys.
1025
1026 my $guard = db_mon mygroup => sub {
1027 my ($family, $a, $c, $d) = @_;
1028 print "mygroup members: ", (join " ", keys %$family), "\n";
1029 };
1030
1031Exmaple: wait until the family "My::Module::workers" is non-empty.
1032
1033 my $guard; $guard = db_mon My::Module::workers => sub {
1034 my ($family, $a, $c, $d) = @_;
1035 return unless %$family;
1036 undef $guard;
1037 print "My::Module::workers now nonempty\n";
1038 };
1039
1040Example: print all changes to the family "AnyRvent::Fantasy::Module".
1041
1042 my $guard = db_mon AnyRvent::Fantasy::Module => sub {
1043 my ($family, $a, $c, $d) = @_;
1044
1045 print "+$_=$family->{$_}\n" for @$a;
1046 print "*$_=$family->{$_}\n" for @$c;
1047 print "-$_=$family->{$_}\n" for @$d;
1048 };
1049
1050=cut
1051
1052=back
1053
879=head1 AnyEvent::MP vs. Distributed Erlang 1054=head1 AnyEvent::MP vs. Distributed Erlang
880 1055
881AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 1056AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
882== aemp node, Erlang process == aemp port), so many of the documents and 1057== aemp node, Erlang process == aemp port), so many of the documents and
883programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 1058programming techniques employed by Erlang apply to AnyEvent::MP. Here is a

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines