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.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
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
267That means that the values specified in the profile have highest priority 279That means that the values specified in the profile have highest priority
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. The 284this 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. 285a unique randoms tring (C</%u>) appended.
286
287The node ID can contain some C<%> sequences that are expanded: C<%n>
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>.
274 292
275=item step 2, bind listener sockets 293=item step 2, bind listener sockets
276 294
277The 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
278aemp protocol listeners on all binds specified (it is possible and valid 296aemp protocol listeners on all binds specified (it is possible and valid
295Example: become a distributed node using the local node name as profile. 313Example: become a distributed node using the local node name as profile.
296This 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.
297 315
298 configure 316 configure
299 317
300Example: become an anonymous node. This form is often used for commandline 318Example: become a semi-anonymous node. This form is often used for
301clients. 319commandline clients.
302 320
303 configure nodeid => "anon/"; 321 configure nodeid => "myscript/%n/%u";
304 322
305Example: configure a node using a profile called seed, which is suitable 323Example: 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, 324for a seed node as it binds on all local addresses on a fixed port (4040,
307customary for aemp). 325customary for aemp).
308 326
309 # use the aemp commandline utility 327 # use the aemp commandline utility
310 # aemp profile seed nodeid anon/ binds '*:4040' 328 # aemp profile seed binds '*:4040'
311 329
312 # then use it 330 # then use it
313 configure profile => "seed"; 331 configure profile => "seed";
314 332
315 # or simply use aemp from the shell again: 333 # or simply use aemp from the shell again:
380 398
381=cut 399=cut
382 400
383sub rcv($@); 401sub rcv($@);
384 402
385sub _kilme { 403my $KILME = sub {
386 die "received message on port without callback"; 404 die "received message on port without callback";
387} 405};
388 406
389sub port(;&) { 407sub port(;&) {
390 my $id = "$UNIQ." . ++$ID; 408 my $id = $UNIQ . ++$ID;
391 my $port = "$NODE#$id"; 409 my $port = "$NODE#$id";
392 410
393 rcv $port, shift || \&_kilme; 411 rcv $port, shift || $KILME;
394 412
395 $port 413 $port
396} 414}
397 415
398=item rcv $local_port, $callback->(@msg) 416=item rcv $local_port, $callback->(@msg)
668 } 686 }
669 687
670 $node->monitor ($port, $cb); 688 $node->monitor ($port, $cb);
671 689
672 defined wantarray 690 defined wantarray
673 and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }) 691 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
674} 692}
675 693
676=item $guard = mon_guard $port, $ref, $ref... 694=item $guard = mon_guard $port, $ref, $ref...
677 695
678Monitors the given C<$port> and keeps the passed references. When the port 696Monitors the given C<$port> and keeps the passed references. When the port
782} 800}
783 801
784sub spawn(@) { 802sub spawn(@) {
785 my ($nodeid, undef) = split /#/, shift, 2; 803 my ($nodeid, undef) = split /#/, shift, 2;
786 804
787 my $id = "$RUNIQ." . ++$ID; 805 my $id = $RUNIQ . ++$ID;
788 806
789 $_[0] =~ /::/ 807 $_[0] =~ /::/
790 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 808 or Carp::croak "spawn init function must be a fully-qualified name, caught";
791 809
792 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_; 810 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
816 ref $action[0] 834 ref $action[0]
817 ? $action[0]() 835 ? $action[0]()
818 : snd @action; 836 : snd @action;
819 }; 837 };
820} 838}
839
840#=item $cb2 = timeout $seconds, $cb[, @args]
821 841
822=item cal $port, @msg, $callback[, $timeout] 842=item cal $port, @msg, $callback[, $timeout]
823 843
824A 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
825given contents (C<@msg>), but adds a reply port to the message. 845given contents (C<@msg>), but adds a reply port to the message.
871 $port 891 $port
872} 892}
873 893
874=back 894=back
875 895
896=head1 DISTRIBUTED DATABASE
897
898AnyEvent::MP comes with a simple distributed database. The database will
899be mirrored asynchronously on all global nodes. Other nodes bind to one
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.
903
904The database structure is that of a two-level hash - the database hash
905contains hashes which contain values, similarly to a perl hash of hashes,
906i.e.:
907
908 $DATABASE{$family}{$subkey} = $value
909
910The top level hash key is called "family", and the second-level hash key
911is called "subkey" or simply "key".
912
913The family must be alphanumeric, i.e. start with a letter and consist
914of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>,
915pretty much like Perl module names.
916
917As the family namespace is global, it is recommended to prefix family names
918with the name of the application or module using it.
919
920The subkeys must be non-empty strings, with no further restrictions.
921
922The values should preferably be strings, but other perl scalars should
923work as well (such as C<undef>, arrays and hashes).
924
925Every database entry is owned by one node - adding the same family/subkey
926combination on multiple nodes will not cause discomfort for AnyEvent::MP,
927but the result might be nondeterministic, i.e. the key might have
928different values on different nodes.
929
930Different subkeys in the same family can be owned by different nodes
931without problems, and in fact, this is the common method to create worker
932pools. For example, a worker port for image scaling might do this:
933
934 db_set my_image_scalers => $port;
935
936And clients looking for an image scaler will want to get the
937C<my_image_scalers> keys from time to time:
938
939 db_keys my_image_scalers => sub {
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.
955
956=over
957
958=item db_set $family => $subkey [=> $value]
959
960Sets (or replaces) a key to the database - if C<$value> is omitted,
961C<undef> is used instead.
962
963=item db_del $family => $subkey...
964
965Deletes one or more subkeys from the database family.
966
967=item $guard = db_reg $family => $subkey [=> $value]
968
969Sets the key on the database and returns a guard. When the guard is
970destroyed, the key is deleted from the database. If C<$value> is missing,
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 };
1038
1039=cut
1040
1041=back
1042
876=head1 AnyEvent::MP vs. Distributed Erlang 1043=head1 AnyEvent::MP vs. Distributed Erlang
877 1044
878AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 1045AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
879== aemp node, Erlang process == aemp port), so many of the documents and 1046== aemp node, Erlang process == aemp port), so many of the documents and
880programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 1047programming techniques employed by Erlang apply to AnyEvent::MP. Here is a

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines