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.120 by root, Sun Feb 26 11:12:54 2012 UTC vs.
Revision 1.136 by root, Wed Mar 21 15:22:16 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
175 176
176=cut 177=cut
177 178
178package AnyEvent::MP; 179package AnyEvent::MP;
179 180
181use AnyEvent::MP::Config ();
180use AnyEvent::MP::Kernel; 182use AnyEvent::MP::Kernel;
183use AnyEvent::MP::Kernel qw(%NODE %PORT %PORT_DATA $UNIQ $RUNIQ $ID);
181 184
182use common::sense; 185use common::sense;
183 186
184use Carp (); 187use Carp ();
185 188
186use AE (); 189use AE ();
190use Guard ();
187 191
188use base "Exporter"; 192use base "Exporter";
189 193
190our $VERSION = '1.30'; 194our $VERSION = $AnyEvent::MP::Config::VERSION;
191 195
192our @EXPORT = qw( 196our @EXPORT = qw(
193 NODE $NODE *SELF node_of after 197 NODE $NODE *SELF node_of after
194 configure 198 configure
195 snd rcv mon mon_guard kil psub peval spawn cal 199 snd rcv mon mon_guard kil psub peval spawn cal
196 port 200 port
201 db_set db_del db_reg
202 db_mon db_family db_keys db_values
197); 203);
198 204
199our $SELF; 205our $SELF;
200 206
201sub _self_die() { 207sub _self_die() {
221Before a node can talk to other nodes on the network (i.e. enter 227Before a node can talk to other nodes on the network (i.e. enter
222"distributed mode") it has to configure itself - the minimum a node needs 228"distributed mode") it has to configure itself - the minimum a node needs
223to know is its own name, and optionally it should know the addresses of 229to know is its own name, and optionally it should know the addresses of
224some other nodes in the network to discover other nodes. 230some other nodes in the network to discover other nodes.
225 231
226The key/value pairs are basically the same ones as documented for the
227F<aemp> command line utility (sans the set/del prefix).
228
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.
234
235The key/value pairs are basically the same ones as documented for the
236F<aemp> command line utility (sans the set/del prefix), with these additions:
237
238=over 4
239
240=item norc => $boolean (default false)
241
242If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not>
243be consulted - all configuraiton options must be specified in the
244C<configure> call.
245
246=item force => $boolean (default false)
247
248IF true, then the values specified in the C<configure> will take
249precedence over any values configured via the rc file. The default is for
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.
264
265=back
231 266
232=over 4 267=over 4
233 268
234=item step 1, gathering configuration from profiles 269=item step 1, gathering configuration from profiles
235 270
249That means that the values specified in the profile have highest priority 284That means that the values specified in the profile have highest priority
250and the values specified directly via C<configure> have lowest priority, 285and the values specified directly via C<configure> have lowest priority,
251and can only be used to specify defaults. 286and can only be used to specify defaults.
252 287
253If 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
254this process. If not, then the profile name will be used as node ID. The 289this process. If not, then the profile name will be used as node ID, with
255special node ID of C<anon/> will be replaced by a random node ID. 290a unique randoms tring (C</%u>) appended.
291
292The node ID can contain some C<%> sequences that are expanded: C<%n>
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>.
256 297
257=item step 2, bind listener sockets 298=item step 2, bind listener sockets
258 299
259The 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
260aemp protocol listeners on all binds specified (it is possible and valid 301aemp protocol listeners on all binds specified (it is possible and valid
277Example: become a distributed node using the local node name as profile. 318Example: become a distributed node using the local node name as profile.
278This 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.
279 320
280 configure 321 configure
281 322
282Example: become an anonymous node. This form is often used for commandline 323Example: become a semi-anonymous node. This form is often used for
283clients. 324commandline clients.
284 325
285 configure nodeid => "anon/"; 326 configure nodeid => "myscript/%n/%u";
286 327
287Example: configure a node using a profile called seed, which is suitable 328Example: configure a node using a profile called seed, which is suitable
288for 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,
289customary for aemp). 330customary for aemp).
290 331
291 # use the aemp commandline utility 332 # use the aemp commandline utility
292 # aemp profile seed nodeid anon/ binds '*:4040' 333 # aemp profile seed binds '*:4040'
293 334
294 # then use it 335 # then use it
295 configure profile => "seed"; 336 configure profile => "seed";
296 337
297 # or simply use aemp from the shell again: 338 # or simply use aemp from the shell again:
362 403
363=cut 404=cut
364 405
365sub rcv($@); 406sub rcv($@);
366 407
367sub _kilme { 408my $KILME = sub {
368 die "received message on port without callback"; 409 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
369} 410 kil $SELF, unhandled_message => "no callback found for message '$tag'";
411};
370 412
371sub port(;&) { 413sub port(;&) {
372 my $id = "$UNIQ." . $ID++; 414 my $id = $UNIQ . ++$ID;
373 my $port = "$NODE#$id"; 415 my $port = "$NODE#$id";
374 416
375 rcv $port, shift || \&_kilme; 417 rcv $port, shift || $KILME;
376 418
377 $port 419 $port
378} 420}
379 421
380=item rcv $local_port, $callback->(@msg) 422=item rcv $local_port, $callback->(@msg)
385 427
386The global C<$SELF> (exported by this module) contains C<$port> while 428The global C<$SELF> (exported by this module) contains C<$port> while
387executing the callback. Runtime errors during callback execution will 429executing the callback. Runtime errors during callback execution will
388result in the port being C<kil>ed. 430result in the port being C<kil>ed.
389 431
390The default callback received all messages not matched by a more specific 432The default callback receives all messages not matched by a more specific
391C<tag> match. 433C<tag> match.
392 434
393=item rcv $local_port, tag => $callback->(@msg_without_tag), ... 435=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
394 436
395Register (or replace) callbacks to be called on messages starting with the 437Register (or replace) callbacks to be called on messages starting with the
650 } 692 }
651 693
652 $node->monitor ($port, $cb); 694 $node->monitor ($port, $cb);
653 695
654 defined wantarray 696 defined wantarray
655 and ($cb += 0, AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }) 697 and ($cb += 0, Guard::guard { $node->unmonitor ($port, $cb) })
656} 698}
657 699
658=item $guard = mon_guard $port, $ref, $ref... 700=item $guard = mon_guard $port, $ref, $ref...
659 701
660Monitors the given C<$port> and keeps the passed references. When the port 702Monitors the given C<$port> and keeps the passed references. When the port
696will be reported as reason C<< die => $@ >>. 738will be reported as reason C<< die => $@ >>.
697 739
698Transport/communication errors are reported as C<< transport_error => 740Transport/communication errors are reported as C<< transport_error =>
699$message >>. 741$message >>.
700 742
701=cut 743Common idioms:
744
745 # silently remove yourself, do not kill linked ports
746 kil $SELF;
747
748 # report a failure in some detail
749 kil $SELF, failure_mode_1 => "it failed with too high temperature";
750
751 # do not waste much time with killing, just die when something goes wrong
752 open my $fh, "<file"
753 or die "file: $!";
702 754
703=item $port = spawn $node, $initfunc[, @initdata] 755=item $port = spawn $node, $initfunc[, @initdata]
704 756
705Creates a port on the node C<$node> (which can also be a port ID, in which 757Creates a port on the node C<$node> (which can also be a port ID, in which
706case it's the node where that port resides). 758case it's the node where that port resides).
764} 816}
765 817
766sub spawn(@) { 818sub spawn(@) {
767 my ($nodeid, undef) = split /#/, shift, 2; 819 my ($nodeid, undef) = split /#/, shift, 2;
768 820
769 my $id = "$RUNIQ." . $ID++; 821 my $id = $RUNIQ . ++$ID;
770 822
771 $_[0] =~ /::/ 823 $_[0] =~ /::/
772 or Carp::croak "spawn init function must be a fully-qualified name, caught"; 824 or Carp::croak "spawn init function must be a fully-qualified name, caught";
773 825
774 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_; 826 snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_;
775 827
776 "$nodeid#$id" 828 "$nodeid#$id"
777} 829}
830
778 831
779=item after $timeout, @msg 832=item after $timeout, @msg
780 833
781=item after $timeout, $callback 834=item after $timeout, $callback
782 835
797 ref $action[0] 850 ref $action[0]
798 ? $action[0]() 851 ? $action[0]()
799 : snd @action; 852 : snd @action;
800 }; 853 };
801} 854}
855
856#=item $cb2 = timeout $seconds, $cb[, @args]
802 857
803=item cal $port, @msg, $callback[, $timeout] 858=item cal $port, @msg, $callback[, $timeout]
804 859
805A simple form of RPC - sends a message to the given C<$port> with the 860A simple form of RPC - sends a message to the given C<$port> with the
806given contents (C<@msg>), but adds a reply port to the message. 861given contents (C<@msg>), but adds a reply port to the message.
852 $port 907 $port
853} 908}
854 909
855=back 910=back
856 911
912=head1 DISTRIBUTED DATABASE
913
914AnyEvent::MP comes with a simple distributed database. The database will
915be mirrored asynchronously on all global nodes. Other nodes bind to one
916of the global nodes for their needs. Every node has a "local database"
917which contains all the values that are set locally. All local databases
918are merged together to form the global database, which can be queried.
919
920The database structure is that of a two-level hash - the database hash
921contains hashes which contain values, similarly to a perl hash of hashes,
922i.e.:
923
924 $DATABASE{$family}{$subkey} = $value
925
926The top level hash key is called "family", and the second-level hash key
927is called "subkey" or simply "key".
928
929The family must be alphanumeric, i.e. start with a letter and consist
930of letters, digits, underscores and colons (C<[A-Za-z][A-Za-z0-9_:]*>,
931pretty much like Perl module names.
932
933As the family namespace is global, it is recommended to prefix family names
934with the name of the application or module using it.
935
936The subkeys must be non-empty strings, with no further restrictions.
937
938The values should preferably be strings, but other perl scalars should
939work as well (such as C<undef>, arrays and hashes).
940
941Every database entry is owned by one node - adding the same family/subkey
942combination on multiple nodes will not cause discomfort for AnyEvent::MP,
943but the result might be nondeterministic, i.e. the key might have
944different values on different nodes.
945
946Different subkeys in the same family can be owned by different nodes
947without problems, and in fact, this is the common method to create worker
948pools. For example, a worker port for image scaling might do this:
949
950 db_set my_image_scalers => $port;
951
952And clients looking for an image scaler will want to get the
953C<my_image_scalers> keys from time to time:
954
955 db_keys my_image_scalers => sub {
956 @ports = @{ $_[0] };
957 };
958
959Or better yet, they want to monitor the database family, so they always
960have a reasonable up-to-date copy:
961
962 db_mon my_image_scalers => sub {
963 @ports = keys %{ $_[0] };
964 };
965
966In general, you can set or delete single subkeys, but query and monitor
967whole families only.
968
969If you feel the need to monitor or query a single subkey, try giving it
970it's own family.
971
972=over
973
974=item db_set $family => $subkey [=> $value]
975
976Sets (or replaces) a key to the database - if C<$value> is omitted,
977C<undef> is used instead.
978
979=item db_del $family => $subkey...
980
981Deletes one or more subkeys from the database family.
982
983=item $guard = db_reg $family => $subkey [=> $value]
984
985Sets the key on the database and returns a guard. When the guard is
986destroyed, the key is deleted from the database. If C<$value> is missing,
987then C<undef> is used.
988
989=item db_family $family => $cb->(\%familyhash)
990
991Queries the named database C<$family> and call the callback with the
992family represented as a hash. You can keep and freely modify the hash.
993
994=item db_keys $family => $cb->(\@keys)
995
996Same as C<db_family>, except it only queries the family I<subkeys> and passes
997them as array reference to the callback.
998
999=item db_values $family => $cb->(\@values)
1000
1001Same as C<db_family>, except it only queries the family I<values> and passes them
1002as array reference to the callback.
1003
1004=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted)
1005
1006Creates a monitor on the given database family. Each time a key is set
1007or or is deleted the callback is called with a hash containing the
1008database family and three lists of added, changed and deleted subkeys,
1009respectively. If no keys have changed then the array reference might be
1010C<undef> or even missing.
1011
1012If not called in void context, a guard object is returned that, when
1013destroyed, stops the monitor.
1014
1015The family hash reference and the key arrays belong to AnyEvent::MP and
1016B<must not be modified or stored> by the callback. When in doubt, make a
1017copy.
1018
1019As soon as possible after the monitoring starts, the callback will be
1020called with the intiial contents of the family, even if it is empty,
1021i.e. there will always be a timely call to the callback with the current
1022contents.
1023
1024It is possible that the callback is called with a change event even though
1025the subkey is already present and the value has not changed.
1026
1027The monitoring stops when the guard object is destroyed.
1028
1029Example: on every change to the family "mygroup", print out all keys.
1030
1031 my $guard = db_mon mygroup => sub {
1032 my ($family, $a, $c, $d) = @_;
1033 print "mygroup members: ", (join " ", keys %$family), "\n";
1034 };
1035
1036Exmaple: wait until the family "My::Module::workers" is non-empty.
1037
1038 my $guard; $guard = db_mon My::Module::workers => sub {
1039 my ($family, $a, $c, $d) = @_;
1040 return unless %$family;
1041 undef $guard;
1042 print "My::Module::workers now nonempty\n";
1043 };
1044
1045Example: print all changes to the family "AnyRvent::Fantasy::Module".
1046
1047 my $guard = db_mon AnyRvent::Fantasy::Module => sub {
1048 my ($family, $a, $c, $d) = @_;
1049
1050 print "+$_=$family->{$_}\n" for @$a;
1051 print "*$_=$family->{$_}\n" for @$c;
1052 print "-$_=$family->{$_}\n" for @$d;
1053 };
1054
1055=cut
1056
1057=back
1058
857=head1 AnyEvent::MP vs. Distributed Erlang 1059=head1 AnyEvent::MP vs. Distributed Erlang
858 1060
859AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node 1061AnyEvent::MP got lots of its ideas from distributed Erlang (Erlang node
860== aemp node, Erlang process == aemp port), so many of the documents and 1062== aemp node, Erlang process == aemp port), so many of the documents and
861programming techniques employed by Erlang apply to AnyEvent::MP. Here is a 1063programming techniques employed by Erlang apply to AnyEvent::MP. Here is a

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines