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.128 by root, Sun Mar 4 14:28:44 2012 UTC vs.
Revision 1.147 by root, Mon Jun 20 16:58:15 2016 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
48 my $timer = AE::timer 1, 0, psub { 48 my $timer = AE::timer 1, 0, psub {
49 die "kill the port, delayed"; 49 die "kill the port, delayed";
50 }; 50 };
51 51
52=head1 CURRENT STATUS 52 # distributed database - modification
53 db_set $family => $subkey [=> $value] # add a subkey
54 db_del $family => $subkey... # delete one or more subkeys
55 db_reg $family => $port [=> $value] # register a port
53 56
54 bin/aemp - stable. 57 # distributed database - queries
55 AnyEvent::MP - stable API, should work. 58 db_family $family => $cb->(\%familyhash)
56 AnyEvent::MP::Intro - explains most concepts. 59 db_keys $family => $cb->(\@keys)
57 AnyEvent::MP::Kernel - mostly stable API. 60 db_values $family => $cb->(\@values)
58 AnyEvent::MP::Global - stable API. 61
62 # distributed database - monitoring a family
63 db_mon $family => $cb->(\%familyhash, \@added, \@changed, \@deleted)
59 64
60=head1 DESCRIPTION 65=head1 DESCRIPTION
61 66
62This module (-family) implements a simple message passing framework. 67This module (-family) implements a simple message passing framework.
63 68
113each other. To do this, nodes should listen on one or more local transport 118each other. To do this, nodes should listen on one or more local transport
114endpoints - binds. 119endpoints - binds.
115 120
116Currently, only standard C<ip:port> specifications can be used, which 121Currently, only standard C<ip:port> specifications can be used, which
117specify TCP ports to listen on. So a bind is basically just a tcp socket 122specify TCP ports to listen on. So a bind is basically just a tcp socket
118in listening mode thta accepts conenctions form other nodes. 123in listening mode that accepts connections from other nodes.
119 124
120=item seed nodes 125=item seed nodes
121 126
122When a node starts, it knows nothing about the network it is in - it 127When a node starts, it knows nothing about the network it is in - it
123needs to connect to at least one other node that is already in the 128needs to connect to at least one other node that is already in the
124network. These other nodes are called "seed nodes". 129network. These other nodes are called "seed nodes".
125 130
126Seed nodes themselves are not special - they are seed nodes only because 131Seed nodes themselves are not special - they are seed nodes only because
127some other node I<uses> them as such, but any node can be used as seed 132some other node I<uses> them as such, but any node can be used as seed
128node for other nodes, and eahc node cna use a different set of seed nodes. 133node for other nodes, and eahc node can use a different set of seed nodes.
129 134
130In addition to discovering the network, seed nodes are also used to 135In addition to discovering the network, seed nodes are also used to
131maintain the network - all nodes using the same seed node form are part of 136maintain the network - all nodes using the same seed node are part of the
132the same network. If a network is split into multiple subnets because e.g. 137same network. If a network is split into multiple subnets because e.g. the
133the network link between the parts goes down, then using the same seed 138network link between the parts goes down, then using the same seed nodes
134nodes for all nodes ensures that eventually the subnets get merged again. 139for all nodes ensures that eventually the subnets get merged again.
135 140
136Seed nodes are expected to be long-running, and at least one seed node 141Seed nodes are expected to be long-running, and at least one seed node
137should always be available. They should also be relatively responsive - a 142should always be available. They should also be relatively responsive - a
138seed node that blocks for long periods will slow down everybody else. 143seed node that blocks for long periods will slow down everybody else.
139 144
163 168
164Any node that loads the L<AnyEvent::MP::Global> module becomes a global 169Any node that loads the L<AnyEvent::MP::Global> module becomes a global
165node and tries to keep connections to all other nodes. So while it can 170node and tries to keep connections to all other nodes. So while it can
166make sense to make every node "global" in small networks, it usually makes 171make sense to make every node "global" in small networks, it usually makes
167sense to only make seed nodes into global nodes in large networks (nodes 172sense to only make seed nodes into global nodes in large networks (nodes
168keep connections to seed nodes and global nodes, so makign them the same 173keep connections to seed nodes and global nodes, so making them the same
169reduces overhead). 174reduces overhead).
170 175
171=back 176=back
172 177
173=head1 VARIABLES/FUNCTIONS 178=head1 VARIABLES/FUNCTIONS
178 183
179package AnyEvent::MP; 184package AnyEvent::MP;
180 185
181use AnyEvent::MP::Config (); 186use AnyEvent::MP::Config ();
182use AnyEvent::MP::Kernel; 187use AnyEvent::MP::Kernel;
183use AnyEvent::MP::Kernel qw(%NODE %PORT %PORT_DATA $UNIQ $RUNIQ $ID); 188use AnyEvent::MP::Kernel qw(
189 %NODE %PORT %PORT_DATA $UNIQ $RUNIQ $ID
190 add_node load_func
191
192 NODE $NODE
193 configure
194 node_of port_is_local
195 snd kil
196 db_set db_del
197 db_mon db_family db_keys db_values
198);
184 199
185use common::sense; 200use common::sense;
186 201
187use Carp (); 202use Carp ();
188 203
189use AE (); 204use AnyEvent ();
190use Guard (); 205use Guard ();
191 206
192use base "Exporter"; 207use base "Exporter";
193 208
194our $VERSION = $AnyEvent::MP::Config::VERSION; 209our $VERSION = $AnyEvent::MP::Config::VERSION;
195 210
196our @EXPORT = qw( 211our @EXPORT = qw(
197 NODE $NODE *SELF node_of after 212 NODE $NODE
198 configure 213 configure
214 node_of port_is_local
215 snd kil
216 db_set db_del
217 db_mon db_family db_keys db_values
218
219 *SELF
220
199 snd rcv mon mon_guard kil psub peval spawn cal 221 port rcv mon mon_guard psub peval spawn cal
200 port
201 db_set db_del db_reg 222 db_set db_del db_reg
202 db_mon db_family db_keys db_values 223 db_mon db_family db_keys db_values
224
225 after
203); 226);
204 227
205our $SELF; 228our $SELF;
206 229
207sub _self_die() { 230sub _self_die() {
218 241
219=item $nodeid = node_of $port 242=item $nodeid = node_of $port
220 243
221Extracts and returns the node ID from a port ID or a node ID. 244Extracts and returns the node ID from a port ID or a node ID.
222 245
246=item $is_local = port_is_local $port
247
248Returns true iff the port is a local port.
249
223=item configure $profile, key => value... 250=item configure $profile, key => value...
224 251
225=item configure key => value... 252=item configure key => value...
226 253
227Before a node can talk to other nodes on the network (i.e. enter 254Before a node can talk to other nodes on the network (i.e. enter
238=over 4 265=over 4
239 266
240=item norc => $boolean (default false) 267=item norc => $boolean (default false)
241 268
242If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not> 269If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not>
243be consulted - all configuraiton options must be specified in the 270be consulted - all configuration options must be specified in the
244C<configure> call. 271C<configure> call.
245 272
246=item force => $boolean (default false) 273=item force => $boolean (default false)
247 274
248IF true, then the values specified in the C<configure> will take 275IF true, then the values specified in the C<configure> will take
249precedence over any values configured via the rc file. The default is for 276precedence over any values configured via the rc file. The default is for
250the rc file to override any options specified in the program. 277the 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.
259 278
260=back 279=back
261 280
262=over 4 281=over 4
263 282
398 417
399=cut 418=cut
400 419
401sub rcv($@); 420sub rcv($@);
402 421
403sub _kilme { 422my $KILME = sub {
404 die "received message on port without callback"; 423 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g;
405} 424 kil $SELF, unhandled_message => "no callback found for message '$tag'";
425};
406 426
407sub port(;&) { 427sub port(;&) {
408 my $id = $UNIQ . ++$ID; 428 my $id = $UNIQ . ++$ID;
409 my $port = "$NODE#$id"; 429 my $port = "$NODE#$id";
410 430
411 rcv $port, shift || \&_kilme; 431 rcv $port, shift || $KILME;
412 432
413 $port 433 $port
414} 434}
415 435
416=item rcv $local_port, $callback->(@msg) 436=item rcv $local_port, $callback->(@msg)
421 441
422The global C<$SELF> (exported by this module) contains C<$port> while 442The global C<$SELF> (exported by this module) contains C<$port> while
423executing the callback. Runtime errors during callback execution will 443executing the callback. Runtime errors during callback execution will
424result in the port being C<kil>ed. 444result in the port being C<kil>ed.
425 445
426The default callback received all messages not matched by a more specific 446The default callback receives all messages not matched by a more specific
427C<tag> match. 447C<tag> match.
428 448
429=item rcv $local_port, tag => $callback->(@msg_without_tag), ... 449=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
430 450
431Register (or replace) callbacks to be called on messages starting with the 451Register (or replace) callbacks to be called on messages starting with the
466 486
467sub rcv($@) { 487sub rcv($@) {
468 my $port = shift; 488 my $port = shift;
469 my ($nodeid, $portid) = split /#/, $port, 2; 489 my ($nodeid, $portid) = split /#/, $port, 2;
470 490
471 $NODE{$nodeid} == $NODE{""} 491 $nodeid eq $NODE
472 or Carp::croak "$port: rcv can only be called on local ports, caught"; 492 or Carp::croak "$port: rcv can only be called on local ports, caught";
473 493
474 while (@_) { 494 while (@_) {
475 if (ref $_[0]) { 495 if (ref $_[0]) {
476 if (my $self = $PORT_DATA{$portid}) { 496 if (my $self = $PORT_DATA{$portid}) {
520} 540}
521 541
522=item peval $port, $coderef[, @args] 542=item peval $port, $coderef[, @args]
523 543
524Evaluates the given C<$codref> within the contetx of C<$port>, that is, 544Evaluates the given C<$codref> within the contetx of C<$port>, that is,
525when the code throews an exception the C<$port> will be killed. 545when the code throws an exception the C<$port> will be killed.
526 546
527Any remaining args will be passed to the callback. Any return values will 547Any remaining args will be passed to the callback. Any return values will
528be returned to the caller. 548be returned to the caller.
529 549
530This is useful when you temporarily want to execute code in the context of 550This is useful when you temporarily want to execute code in the context of
595 $res 615 $res
596 } 616 }
597 } 617 }
598} 618}
599 619
620=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
621
622=item $guard = mon $port # kill $SELF when $port dies
623
600=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies 624=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
601
602=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
603
604=item $guard = mon $port # kill $SELF when $port dies
605 625
606=item $guard = mon $port, $rcvport, @msg # send a message when $port dies 626=item $guard = mon $port, $rcvport, @msg # send a message when $port dies
607 627
608Monitor the given port and do something when the port is killed or 628Monitor the given port and do something when the port is killed or
609messages to it were lost, and optionally return a guard that can be used 629messages to it were lost, and optionally return a guard that can be used
610to stop monitoring again. 630to stop monitoring again.
611 631
632The first two forms distinguish between "normal" and "abnormal" kil's:
633
634In the first form (another port given), if the C<$port> is C<kil>'ed with
635a non-empty reason, the other port (C<$rcvport>) will be kil'ed with the
636same reason. That is, on "normal" kil's nothing happens, while under all
637other conditions, the other port is killed with the same reason.
638
639The second form (kill self) is the same as the first form, except that
640C<$rvport> defaults to C<$SELF>.
641
642The remaining forms don't distinguish between "normal" and "abnormal" kil's
643- it's up to the callback or receiver to check whether the C<@reason> is
644empty and act accordingly.
645
612In the first form (callback), the callback is simply called with any 646In the third form (callback), the callback is simply called with any
613number of C<@reason> elements (no @reason means that the port was deleted 647number of C<@reason> elements (empty @reason means that the port was deleted
614"normally"). Note also that I<< the callback B<must> never die >>, so use 648"normally"). Note also that I<< the callback B<must> never die >>, so use
615C<eval> if unsure. 649C<eval> if unsure.
616 650
617In the second form (another port given), the other port (C<$rcvport>)
618will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
619"normal" kils nothing happens, while under all other conditions, the other
620port is killed with the same reason.
621
622The third form (kill self) is the same as the second form, except that
623C<$rvport> defaults to C<$SELF>.
624
625In the last form (message), a message of the form C<@msg, @reason> will be 651In the last form (message), a message of the form C<$rcvport, @msg,
626C<snd>. 652@reason> will be C<snd>.
627 653
628Monitoring-actions are one-shot: once messages are lost (and a monitoring 654Monitoring-actions are one-shot: once messages are lost (and a monitoring
629alert was raised), they are removed and will not trigger again. 655alert was raised), they are removed and will not trigger again, even if it
656turns out that the port is still alive.
630 657
631As a rule of thumb, monitoring requests should always monitor a port from 658As a rule of thumb, monitoring requests should always monitor a remote
632a local port (or callback). The reason is that kill messages might get 659port locally (using a local C<$rcvport> or a callback). The reason is that
633lost, just like any other message. Another less obvious reason is that 660kill messages might get lost, just like any other message. Another less
634even monitoring requests can get lost (for example, when the connection 661obvious reason is that even monitoring requests can get lost (for example,
635to the other node goes down permanently). When monitoring a port locally 662when the connection to the other node goes down permanently). When
636these problems do not exist. 663monitoring a port locally these problems do not exist.
637 664
638C<mon> effectively guarantees that, in the absence of hardware failures, 665C<mon> effectively guarantees that, in the absence of hardware failures,
639after starting the monitor, either all messages sent to the port will 666after starting the monitor, either all messages sent to the port will
640arrive, or the monitoring action will be invoked after possible message 667arrive, or the monitoring action will be invoked after possible message
641loss has been detected. No messages will be lost "in between" (after 668loss has been detected. No messages will be lost "in between" (after
732will be reported as reason C<< die => $@ >>. 759will be reported as reason C<< die => $@ >>.
733 760
734Transport/communication errors are reported as C<< transport_error => 761Transport/communication errors are reported as C<< transport_error =>
735$message >>. 762$message >>.
736 763
737=cut 764Common idioms:
765
766 # silently remove yourself, do not kill linked ports
767 kil $SELF;
768
769 # report a failure in some detail
770 kil $SELF, failure_mode_1 => "it failed with too high temperature";
771
772 # do not waste much time with killing, just die when something goes wrong
773 open my $fh, "<file"
774 or die "file: $!";
738 775
739=item $port = spawn $node, $initfunc[, @initdata] 776=item $port = spawn $node, $initfunc[, @initdata]
740 777
741Creates a port on the node C<$node> (which can also be a port ID, in which 778Creates a port on the node C<$node> (which can also be a port ID, in which
742case it's the node where that port resides). 779case it's the node where that port resides).
834 ref $action[0] 871 ref $action[0]
835 ? $action[0]() 872 ? $action[0]()
836 : snd @action; 873 : snd @action;
837 }; 874 };
838} 875}
876
877#=item $cb2 = timeout $seconds, $cb[, @args]
839 878
840=item cal $port, @msg, $callback[, $timeout] 879=item cal $port, @msg, $callback[, $timeout]
841 880
842A simple form of RPC - sends a message to the given C<$port> with the 881A simple form of RPC - sends a message to the given C<$port> with the
843given contents (C<@msg>), but adds a reply port to the message. 882given contents (C<@msg>), but adds a reply port to the message.
892=back 931=back
893 932
894=head1 DISTRIBUTED DATABASE 933=head1 DISTRIBUTED DATABASE
895 934
896AnyEvent::MP comes with a simple distributed database. The database will 935AnyEvent::MP comes with a simple distributed database. The database will
897be mirrored asynchronously at all global nodes. Other nodes bind to one of 936be mirrored asynchronously on all global nodes. Other nodes bind to one
898the global nodes for their needs. 937of the global nodes for their needs. Every node has a "local database"
938which contains all the values that are set locally. All local databases
939are merged together to form the global database, which can be queried.
899 940
900The database consists of a two-level hash - a hash contains a hash which 941The database structure is that of a two-level hash - the database hash
901contains values. 942contains hashes which contain values, similarly to a perl hash of hashes,
943i.e.:
944
945 $DATABASE{$family}{$subkey} = $value
902 946
903The top level hash key is called "family", and the second-level hash key 947The top level hash key is called "family", and the second-level hash key
904is called "subkey" or simply "key". 948is called "subkey" or simply "key".
905 949
906The family must be alphanumeric, i.e. start with a letter and consist 950The family must be alphanumeric, i.e. start with a letter and consist
911with the name of the application or module using it. 955with the name of the application or module using it.
912 956
913The subkeys must be non-empty strings, with no further restrictions. 957The subkeys must be non-empty strings, with no further restrictions.
914 958
915The values should preferably be strings, but other perl scalars should 959The values should preferably be strings, but other perl scalars should
916work as well (such as undef, arrays and hashes). 960work as well (such as C<undef>, arrays and hashes).
917 961
918Every database entry is owned by one node - adding the same family/subkey 962Every database entry is owned by one node - adding the same family/subkey
919combination on multiple nodes will not cause discomfort for AnyEvent::MP, 963combination on multiple nodes will not cause discomfort for AnyEvent::MP,
920but the result might be nondeterministic, i.e. the key might have 964but the result might be nondeterministic, i.e. the key might have
921different values on different nodes. 965different values on different nodes.
925pools. For example, a worker port for image scaling might do this: 969pools. For example, a worker port for image scaling might do this:
926 970
927 db_set my_image_scalers => $port; 971 db_set my_image_scalers => $port;
928 972
929And clients looking for an image scaler will want to get the 973And clients looking for an image scaler will want to get the
930C<my_image_scalers> keys: 974C<my_image_scalers> keys from time to time:
931 975
932 db_keys "my_image_scalers" => 60 => sub { 976 db_keys my_image_scalers => sub {
933 #d##TODO# 977 @ports = @{ $_[0] };
978 };
979
980Or better yet, they want to monitor the database family, so they always
981have a reasonable up-to-date copy:
982
983 db_mon my_image_scalers => sub {
984 @ports = keys %{ $_[0] };
985 };
986
987In general, you can set or delete single subkeys, but query and monitor
988whole families only.
989
990If you feel the need to monitor or query a single subkey, try giving it
991it's own family.
934 992
935=over 993=over
936 994
937=item db_set $family => $subkey [=> $value] 995=item $guard = db_set $family => $subkey [=> $value]
938 996
939Sets (or replaces) a key to the database - if C<$value> is omitted, 997Sets (or replaces) a key to the database - if C<$value> is omitted,
940C<undef> is used instead. 998C<undef> is used instead.
941 999
1000When called in non-void context, C<db_set> returns a guard that
1001automatically calls C<db_del> when it is destroyed.
1002
942=item db_del $family => $subkey 1003=item db_del $family => $subkey...
943 1004
944Deletes a key from the database. 1005Deletes one or more subkeys from the database family.
945 1006
946=item $guard = db_reg $family => $subkey [=> $value] 1007=item $guard = db_reg $family => $port => $value
947 1008
948Sets the key on the database and returns a guard. When the guard is 1009=item $guard = db_reg $family => $port
949destroyed, the key is deleted from the database. If C<$value> is missing,
950then C<undef> is used.
951 1010
952=item $guard = db_mon $family => $cb->($familyhash, \@subkeys...) 1011=item $guard = db_reg $family
953 1012
1013Registers a port in the given family and optionally returns a guard to
1014remove it.
1015
1016This function basically does the same as:
1017
1018 db_set $family => $port => $value
1019
1020Except that the port is monitored and automatically removed from the
1021database family when it is kil'ed.
1022
1023If C<$value> is missing, C<undef> is used. If C<$port> is missing, then
1024C<$SELF> is used.
1025
1026This function is most useful to register a port in some port group (which
1027is just another name for a database family), and have it removed when the
1028port is gone. This works best when the port is a local port.
1029
1030=cut
1031
1032sub db_reg($$;$) {
1033 my $family = shift;
1034 my $port = @_ ? shift : $SELF;
1035
1036 my $clr = sub { db_del $family => $port };
1037 mon $port, $clr;
1038
1039 db_set $family => $port => $_[0];
1040
1041 defined wantarray
1042 and &Guard::guard ($clr)
1043}
1044
1045=item db_family $family => $cb->(\%familyhash)
1046
1047Queries the named database C<$family> and call the callback with the
1048family represented as a hash. You can keep and freely modify the hash.
1049
1050=item db_keys $family => $cb->(\@keys)
1051
1052Same as C<db_family>, except it only queries the family I<subkeys> and passes
1053them as array reference to the callback.
1054
1055=item db_values $family => $cb->(\@values)
1056
1057Same as C<db_family>, except it only queries the family I<values> and passes them
1058as array reference to the callback.
1059
1060=item $guard = db_mon $family => $cb->(\%familyhash, \@added, \@changed, \@deleted)
1061
954Creates a monitor on the given database family. Each time a key is set or 1062Creates a monitor on the given database family. Each time a key is
955or is deleted the callback is called with a hash containing the database 1063set or is deleted the callback is called with a hash containing the
956family and an arrayref with subkeys that have changed. 1064database family and three lists of added, changed and deleted subkeys,
1065respectively. If no keys have changed then the array reference might be
1066C<undef> or even missing.
957 1067
958Specifically, if one of the passed subkeys exists in the $familyhash, then 1068If not called in void context, a guard object is returned that, when
959it is currently set to the value in the $familyhash. Otherwise, it has 1069destroyed, stops the monitor.
960been deleted.
961 1070
962The first call will be with the current contents of the family and all 1071The family hash reference and the key arrays belong to AnyEvent::MP and
963keys, as if they were just added. 1072B<must not be modified or stored> by the callback. When in doubt, make a
1073copy.
1074
1075As soon as possible after the monitoring starts, the callback will be
1076called with the intiial contents of the family, even if it is empty,
1077i.e. there will always be a timely call to the callback with the current
1078contents.
964 1079
965It is possible that the callback is called with a change event even though 1080It is possible that the callback is called with a change event even though
966the subkey is already present and the value has not changed. 1081the subkey is already present and the value has not changed.
967 1082
968The monitoring stops when the guard object is destroyed. 1083The monitoring stops when the guard object is destroyed.
969 1084
970Example: on every change to the family "mygroup", print out all keys. 1085Example: on every change to the family "mygroup", print out all keys.
971 1086
972 my $guard = db_mon mygroup => sub { 1087 my $guard = db_mon mygroup => sub {
973 my ($family, $keys) = @_; 1088 my ($family, $a, $c, $d) = @_;
974 print "mygroup members: ", (join " ", keys %$family), "\n"; 1089 print "mygroup members: ", (join " ", keys %$family), "\n";
975 }; 1090 };
976 1091
977Exmaple: wait until the family "My::Module::workers" is non-empty. 1092Exmaple: wait until the family "My::Module::workers" is non-empty.
978 1093
979 my $guard; $guard = db_mon My::Module::workers => sub { 1094 my $guard; $guard = db_mon My::Module::workers => sub {
980 my ($family, $keys) = @_; 1095 my ($family, $a, $c, $d) = @_;
981 return unless %$family; 1096 return unless %$family;
982 undef $guard; 1097 undef $guard;
983 print "My::Module::workers now nonempty\n"; 1098 print "My::Module::workers now nonempty\n";
984 }; 1099 };
985 1100
986Example: print all changes to the family "AnyRvent::Fantasy::Module". 1101Example: print all changes to the family "AnyEvent::Fantasy::Module".
987 1102
988 my $guard = db_mon AnyRvent::Fantasy::Module => sub { 1103 my $guard = db_mon AnyEvent::Fantasy::Module => sub {
989 my ($family, $keys) = @_; 1104 my ($family, $a, $c, $d) = @_;
990 1105
991 for (@$keys) { 1106 print "+$_=$family->{$_}\n" for @$a;
992 print "$_: ", 1107 print "*$_=$family->{$_}\n" for @$c;
993 (exists $family->{$_} 1108 print "-$_=$family->{$_}\n" for @$d;
994 ? $family->{$_}
995 : "(deleted)"),
996 "\n";
997 }
998 }; 1109 };
999 1110
1000=cut 1111=cut
1001 1112
1002=back 1113=back
1048filter messages without dequeuing them. 1159filter messages without dequeuing them.
1049 1160
1050This is not a philosophical difference, but simply stems from AnyEvent::MP 1161This is not a philosophical difference, but simply stems from AnyEvent::MP
1051being event-based, while Erlang is process-based. 1162being event-based, while Erlang is process-based.
1052 1163
1053You cna have a look at L<Coro::MP> for a more Erlang-like process model on 1164You can have a look at L<Coro::MP> for a more Erlang-like process model on
1054top of AEMP and Coro threads. 1165top of AEMP and Coro threads.
1055 1166
1056=item * Erlang sends are synchronous, AEMP sends are asynchronous. 1167=item * Erlang sends are synchronous, AEMP sends are asynchronous.
1057 1168
1058Sending messages in Erlang is synchronous and blocks the process until 1169Sending messages in Erlang is synchronous and blocks the process until
1059a conenction has been established and the message sent (and so does not 1170a connection has been established and the message sent (and so does not
1060need a queue that can overflow). AEMP sends return immediately, connection 1171need a queue that can overflow). AEMP sends return immediately, connection
1061establishment is handled in the background. 1172establishment is handled in the background.
1062 1173
1063=item * Erlang suffers from silent message loss, AEMP does not. 1174=item * Erlang suffers from silent message loss, AEMP does not.
1064 1175
1162Keeping your messages simple, concentrating on data structures rather than 1273Keeping your messages simple, concentrating on data structures rather than
1163objects, will keep your messages clean, tidy and efficient. 1274objects, will keep your messages clean, tidy and efficient.
1164 1275
1165=back 1276=back
1166 1277
1278=head1 PORTING FROM AnyEvent::MP VERSION 1.X
1279
1280AEMP version 2 has a few major incompatible changes compared to version 1:
1281
1282=over 4
1283
1284=item AnyEvent::MP::Global no longer has group management functions.
1285
1286At least not officially - the grp_* functions are still exported and might
1287work, but they will be removed in some later release.
1288
1289AnyEvent::MP now comes with a distributed database that is more
1290powerful. Its database families map closely to port groups, but the API
1291has changed (the functions are also now exported by AnyEvent::MP). Here is
1292a rough porting guide:
1293
1294 grp_reg $group, $port # old
1295 db_reg $group, $port # new
1296
1297 $list = grp_get $group # old
1298 db_keys $group, sub { my $list = shift } # new
1299
1300 grp_mon $group, $cb->(\@ports, $add, $del) # old
1301 db_mon $group, $cb->(\%ports, $add, $change, $del) # new
1302
1303C<grp_reg> is a no-brainer (just replace by C<db_reg>), but C<grp_get> is
1304no longer instant, because the local node might not have a copy of the
1305group. You can either modify your code to allow for a callback, or use
1306C<db_mon> to keep an updated copy of the group:
1307
1308 my $local_group_copy;
1309 db_mon $group => sub { $local_group_copy = $_[0] };
1310
1311 # now "keys %$local_group_copy" always returns the most up-to-date
1312 # list of ports in the group.
1313
1314C<grp_mon> can be replaced by C<db_mon> with minor changes - C<db_mon>
1315passes a hash as first argument, and an extra C<$chg> argument that can be
1316ignored:
1317
1318 db_mon $group => sub {
1319 my ($ports, $add, $chg, $lde) = @_;
1320 $ports = [keys %$ports];
1321
1322 # now $ports, $add and $del are the same as
1323 # were originally passed by grp_mon.
1324 ...
1325 };
1326
1327=item Nodes not longer connect to all other nodes.
1328
1329In AEMP 1.x, every node automatically loads the L<AnyEvent::MP::Global>
1330module, which in turn would create connections to all other nodes in the
1331network (helped by the seed nodes).
1332
1333In version 2.x, global nodes still connect to all other global nodes, but
1334other nodes don't - now every node either is a global node itself, or
1335attaches itself to another global node.
1336
1337If a node isn't a global node itself, then it attaches itself to one
1338of its seed nodes. If that seed node isn't a global node yet, it will
1339automatically be upgraded to a global node.
1340
1341So in many cases, nothing needs to be changed - one just has to make sure
1342that all seed nodes are meshed together with the other seed nodes (as with
1343AEMP 1.x), and other nodes specify them as seed nodes. This is most easily
1344achieved by specifying the same set of seed nodes for all nodes in the
1345network.
1346
1347Not opening a connection to every other node is usually an advantage,
1348except when you need the lower latency of an already established
1349connection. To ensure a node establishes a connection to another node,
1350you can monitor the node port (C<mon $node, ...>), which will attempt to
1351create the connection (and notify you when the connection fails).
1352
1353=item Listener-less nodes (nodes without binds) are gone.
1354
1355And are not coming back, at least not in their old form. If no C<binds>
1356are specified for a node, AnyEvent::MP assumes a default of C<*:*>.
1357
1358There are vague plans to implement some form of routing domains, which
1359might or might not bring back listener-less nodes, but don't count on it.
1360
1361The fact that most connections are now optional somewhat mitigates this,
1362as a node can be effectively unreachable from the outside without any
1363problems, as long as it isn't a global node and only reaches out to other
1364nodes (as opposed to being contacted from other nodes).
1365
1366=item $AnyEvent::MP::Kernel::WARN has gone.
1367
1368AnyEvent has acquired a logging framework (L<AnyEvent::Log>), and AEMP now
1369uses this, and so should your programs.
1370
1371Every module now documents what kinds of messages it generates, with
1372AnyEvent::MP acting as a catch all.
1373
1374On the positive side, this means that instead of setting
1375C<PERL_ANYEVENT_MP_WARNLEVEL>, you can get away by setting C<AE_VERBOSE> -
1376much less to type.
1377
1378=back
1379
1380=head1 LOGGING
1381
1382AnyEvent::MP does not normally log anything by itself, but since it is the
1383root of the contetx hierarchy for AnyEvent::MP modules, it will receive
1384all log messages by submodules.
1385
1167=head1 SEE ALSO 1386=head1 SEE ALSO
1168 1387
1169L<AnyEvent::MP::Intro> - a gentle introduction. 1388L<AnyEvent::MP::Intro> - a gentle introduction.
1170 1389
1171L<AnyEvent::MP::Kernel> - more, lower-level, stuff. 1390L<AnyEvent::MP::Kernel> - more, lower-level, stuff.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines