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.127 by root, Sat Mar 3 20:35:10 2012 UTC vs.
Revision 1.132 by root, Sat Mar 10 20:34:11 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
… …
197 NODE $NODE *SELF node_of after 197 NODE $NODE *SELF node_of after
198 configure 198 configure
199 snd rcv mon mon_guard kil psub peval spawn cal 199 snd rcv mon mon_guard kil psub peval spawn cal
200 port 200 port
201 db_set db_del db_reg 201 db_set db_del db_reg
202 db_mon db_family db_keys db_values
202); 203);
203 204
204our $SELF; 205our $SELF;
205 206
206sub _self_die() { 207sub _self_die() {
… …
397 398
398=cut 399=cut
399 400
400sub rcv($@); 401sub rcv($@);
401 402
402sub _kilme { 403my $KILME = sub {
403 die "received message on port without callback"; 404 die "received message on port without callback";
404} 405};
405 406
406sub port(;&) { 407sub port(;&) {
407 my $id = $UNIQ . ++$ID; 408 my $id = $UNIQ . ++$ID;
408 my $port = "$NODE#$id"; 409 my $port = "$NODE#$id";
409 410
410 rcv $port, shift || \&_kilme; 411 rcv $port, shift || $KILME;
411 412
412 $port 413 $port
413} 414}
414 415
415=item rcv $local_port, $callback->(@msg) 416=item rcv $local_port, $callback->(@msg)
… …
833 ref $action[0] 834 ref $action[0]
834 ? $action[0]() 835 ? $action[0]()
835 : snd @action; 836 : snd @action;
836 }; 837 };
837} 838}
839
840#=item $cb2 = timeout $seconds, $cb[, @args]
838 841
839=item cal $port, @msg, $callback[, $timeout] 842=item cal $port, @msg, $callback[, $timeout]
840 843
841A simple form of RPC - sends a message to the given C<$port> with the 844A simple form of RPC - sends a message to the given C<$port> with the
842given contents (C<@msg>), but adds a reply port to the message. 845given contents (C<@msg>), but adds a reply port to the message.
… …
891=back 894=back
892 895
893=head1 DISTRIBUTED DATABASE 896=head1 DISTRIBUTED DATABASE
894 897
895AnyEvent::MP comes with a simple distributed database. The database will 898AnyEvent::MP comes with a simple distributed database. The database will
896be mirrored asynchronously at all global nodes. Other nodes bind to one of 899be mirrored asynchronously on all global nodes. Other nodes bind to one
897the global nodes for their needs. 900of the global nodes for their needs. Every node has a "local database"
901which contains all the values that are set locally. All local databases
902are merged together to form the global database, which can be queried.
898 903
899The database consists of a two-level hash - a hash contains a hash which 904The database structure is that of a two-level hash - the database hash
900contains values. 905contains hashes which contain values, similarly to a perl hash of hashes,
906i.e.:
907
908 $DATABASE{$family}{$subkey} = $value
901 909
902The top level hash key is called "family", and the second-level hash key 910The top level hash key is called "family", and the second-level hash key
903is called "subkey" or simply "key". 911is called "subkey" or simply "key".
904 912
905The family must be alphanumeric, i.e. start with a letter and consist 913The family must be alphanumeric, i.e. start with a letter and consist
… …
910with the name of the application or module using it. 918with the name of the application or module using it.
911 919
912The subkeys must be non-empty strings, with no further restrictions. 920The subkeys must be non-empty strings, with no further restrictions.
913 921
914The values should preferably be strings, but other perl scalars should 922The values should preferably be strings, but other perl scalars should
915work as well (such as undef, arrays and hashes). 923work as well (such as C<undef>, arrays and hashes).
916 924
917Every database entry is owned by one node - adding the same family/subkey 925Every database entry is owned by one node - adding the same family/subkey
918combination on multiple nodes will not cause discomfort for AnyEvent::MP, 926combination on multiple nodes will not cause discomfort for AnyEvent::MP,
919but the result might be nondeterministic, i.e. the key might have 927but the result might be nondeterministic, i.e. the key might have
920different values on different nodes. 928different values on different nodes.
… …
924pools. For example, a worker port for image scaling might do this: 932pools. For example, a worker port for image scaling might do this:
925 933
926 db_set my_image_scalers => $port; 934 db_set my_image_scalers => $port;
927 935
928And clients looking for an image scaler will want to get the 936And clients looking for an image scaler will want to get the
929C<my_image_scalers> keys: 937C<my_image_scalers> keys from time to time:
930 938
931 db_keys "my_image_scalers" => 60 => sub { 939 db_keys my_image_scalers => sub {
932 #d##TODO# 940 @ports = @{ $_[0] };
941 };
942
943Or better yet, they want to monitor the database family, so they always
944have a reasonable up-to-date copy:
945
946 db_mon my_image_scalers => sub {
947 @ports = keys %{ $_[0] };
948 };
949
950In general, you can set or delete single subkeys, but query and monitor
951whole families only.
952
953If you feel the need to monitor or query a single subkey, try giving it
954it's own family.
933 955
934=over 956=over
935 957
936=item db_set $family => $subkey [=> $value] 958=item db_set $family => $subkey [=> $value]
937 959
938Sets (or replaces) a key to the database - if C<$value> is omitted, 960Sets (or replaces) a key to the database - if C<$value> is omitted,
939C<undef> is used instead. 961C<undef> is used instead.
940 962
941=item db_del $family => $subkey 963=item db_del $family => $subkey...
942 964
943Deletes a key from the database. 965Deletes one or more subkeys from the database family.
944 966
945=item $guard = db_reg $family => $subkey [=> $value] 967=item $guard = db_reg $family => $subkey [=> $value]
946 968
947Sets the key on the database and returns a guard. When the guard is 969Sets the key on the database and returns a guard. When the guard is
948destroyed, the key is deleted from the database. If C<$value> is missing, 970destroyed, the key is deleted from the database. If C<$value> is missing,
949then C<undef> is used. 971then C<undef> is used.
972
973=item db_family $family => $cb->(\%familyhash)
974
975Queries the named database C<$family> and call the callback with the
976family represented as a hash. You can keep and freely modify the hash.
977
978=item db_keys $family => $cb->(\@keys)
979
980Same as C<db_family>, except it only queries the family I<subkeys> and passes
981them as array reference to the callback.
982
983=item db_values $family => $cb->(\@values)
984
985Same as C<db_family>, except it only queries the family I<values> and passes them
986as array reference to the callback.
987
988=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
989
990Creates a monitor on the given database family. Each time a key is set
991or or is deleted the callback is called with a hash containing the
992database family and three lists of added, changed and deleted subkeys,
993respectively. If no keys have changed then the array reference might be
994C<undef> or even missing.
995
996If not called in void context, a guard object is returned that, when
997destroyed, stops the monitor.
998
999The family hash reference and the key arrays belong to AnyEvent::MP and
1000B<must not be modified or stored> by the callback. When in doubt, make a
1001copy.
1002
1003As soon as possible after the monitoring starts, the callback will be
1004called with the intiial contents of the family, even if it is empty,
1005i.e. there will always be a timely call to the callback with the current
1006contents.
1007
1008It is possible that the callback is called with a change event even though
1009the subkey is already present and the value has not changed.
1010
1011The monitoring stops when the guard object is destroyed.
1012
1013Example: on every change to the family "mygroup", print out all keys.
1014
1015 my $guard = db_mon mygroup => sub {
1016 my ($family, $a, $c, $d) = @_;
1017 print "mygroup members: ", (join " ", keys %$family), "\n";
1018 };
1019
1020Exmaple: wait until the family "My::Module::workers" is non-empty.
1021
1022 my $guard; $guard = db_mon My::Module::workers => sub {
1023 my ($family, $a, $c, $d) = @_;
1024 return unless %$family;
1025 undef $guard;
1026 print "My::Module::workers now nonempty\n";
1027 };
1028
1029Example: print all changes to the family "AnyRvent::Fantasy::Module".
1030
1031 my $guard = db_mon AnyRvent::Fantasy::Module => sub {
1032 my ($family, $a, $c, $d) = @_;
1033
1034 print "+$_=$family->{$_}\n" for @$a;
1035 print "*$_=$family->{$_}\n" for @$c;
1036 print "-$_=$family->{$_}\n" for @$d;
1037 };
950 1038
951=cut 1039=cut
952 1040
953=back 1041=back
954 1042

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines