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.138 by root, Thu Mar 22 00:48:29 2012 UTC vs.
Revision 1.148 by root, Mon Jun 20 16:58:30 2016 UTC

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->(@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 278
265=back 279=back
266 280
267=over 4 281=over 4
268 282
297 311
298=item step 2, bind listener sockets 312=item step 2, bind listener sockets
299 313
300The next step is to look up the binds in the profile, followed by binding 314The next step is to look up the binds in the profile, followed by binding
301aemp protocol listeners on all binds specified (it is possible and valid 315aemp protocol listeners on all binds specified (it is possible and valid
302to have no binds, meaning that the node cannot be contacted form the 316to have no binds, meaning that the node cannot be contacted from the
303outside. This means the node cannot talk to other nodes that also have no 317outside. This means the node cannot talk to other nodes that also have no
304binds, but it can still talk to all "normal" nodes). 318binds, but it can still talk to all "normal" nodes).
305 319
306If the profile does not specify a binds list, then a default of C<*> is 320If the profile does not specify a binds list, then a default of C<*> is
307used, meaning the node will bind on a dynamically-assigned port on every 321used, meaning the node will bind on a dynamically-assigned port on every
472 486
473sub rcv($@) { 487sub rcv($@) {
474 my $port = shift; 488 my $port = shift;
475 my ($nodeid, $portid) = split /#/, $port, 2; 489 my ($nodeid, $portid) = split /#/, $port, 2;
476 490
477 $NODE{$nodeid} == $NODE{""} 491 $nodeid eq $NODE
478 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";
479 493
480 while (@_) { 494 while (@_) {
481 if (ref $_[0]) { 495 if (ref $_[0]) {
482 if (my $self = $PORT_DATA{$portid}) { 496 if (my $self = $PORT_DATA{$portid}) {
526} 540}
527 541
528=item peval $port, $coderef[, @args] 542=item peval $port, $coderef[, @args]
529 543
530Evaluates 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,
531when the code throews an exception the C<$port> will be killed. 545when the code throws an exception the C<$port> will be killed.
532 546
533Any 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
534be returned to the caller. 548be returned to the caller.
535 549
536This 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
601 $res 615 $res
602 } 616 }
603 } 617 }
604} 618}
605 619
620=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
621
622=item $guard = mon $port # kill $SELF when $port dies
623
606=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies 624=item $guard = mon $port, $cb->(@reason) # call $cb when $port dies
607
608=item $guard = mon $port, $rcvport # kill $rcvport when $port dies
609
610=item $guard = mon $port # kill $SELF when $port dies
611 625
612=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
613 627
614Monitor 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
615messages 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
616to stop monitoring again. 630to stop monitoring again.
617 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
618In the first form (callback), the callback is simply called with any 646In the third form (callback), the callback is simply called with any
619number 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
620"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
621C<eval> if unsure. 649C<eval> if unsure.
622 650
623In the second form (another port given), the other port (C<$rcvport>)
624will be C<kil>'ed with C<@reason>, if a @reason was specified, i.e. on
625"normal" kils nothing happens, while under all other conditions, the other
626port is killed with the same reason.
627
628The third form (kill self) is the same as the second form, except that
629C<$rvport> defaults to C<$SELF>.
630
631In 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,
632C<snd>. 652@reason> will be C<snd>.
633 653
634Monitoring-actions are one-shot: once messages are lost (and a monitoring 654Monitoring-actions are one-shot: once messages are lost (and a monitoring
635alert 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.
636 657
637As a rule of thumb, monitoring requests should always monitor a port from 658As a rule of thumb, monitoring requests should always monitor a remote
638a 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
639lost, just like any other message. Another less obvious reason is that 660kill messages might get lost, just like any other message. Another less
640even monitoring requests can get lost (for example, when the connection 661obvious reason is that even monitoring requests can get lost (for example,
641to the other node goes down permanently). When monitoring a port locally 662when the connection to the other node goes down permanently). When
642these problems do not exist. 663monitoring a port locally these problems do not exist.
643 664
644C<mon> effectively guarantees that, in the absence of hardware failures, 665C<mon> effectively guarantees that, in the absence of hardware failures,
645after starting the monitor, either all messages sent to the port will 666after starting the monitor, either all messages sent to the port will
646arrive, or the monitoring action will be invoked after possible message 667arrive, or the monitoring action will be invoked after possible message
647loss has been detected. No messages will be lost "in between" (after 668loss has been detected. No messages will be lost "in between" (after
1034=item db_values $family => $cb->(\@values) 1055=item db_values $family => $cb->(\@values)
1035 1056
1036Same as C<db_family>, except it only queries the family I<values> and passes them 1057Same as C<db_family>, except it only queries the family I<values> and passes them
1037as array reference to the callback. 1058as array reference to the callback.
1038 1059
1039=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted) 1060=item $guard = db_mon $family => $cb->(\%familyhash, \@added, \@changed, \@deleted)
1040 1061
1041Creates a monitor on the given database family. Each time a key is set 1062Creates a monitor on the given database family. Each time a key is
1042or or is deleted the callback is called with a hash containing the 1063set or is deleted the callback is called with a hash containing the
1043database family and three lists of added, changed and deleted subkeys, 1064database family and three lists of added, changed and deleted subkeys,
1044respectively. If no keys have changed then the array reference might be 1065respectively. If no keys have changed then the array reference might be
1045C<undef> or even missing. 1066C<undef> or even missing.
1046 1067
1047If not called in void context, a guard object is returned that, when 1068If not called in void context, a guard object is returned that, when
1075 return unless %$family; 1096 return unless %$family;
1076 undef $guard; 1097 undef $guard;
1077 print "My::Module::workers now nonempty\n"; 1098 print "My::Module::workers now nonempty\n";
1078 }; 1099 };
1079 1100
1080Example: print all changes to the family "AnyRvent::Fantasy::Module". 1101Example: print all changes to the family "AnyEvent::Fantasy::Module".
1081 1102
1082 my $guard = db_mon AnyRvent::Fantasy::Module => sub { 1103 my $guard = db_mon AnyEvent::Fantasy::Module => sub {
1083 my ($family, $a, $c, $d) = @_; 1104 my ($family, $a, $c, $d) = @_;
1084 1105
1085 print "+$_=$family->{$_}\n" for @$a; 1106 print "+$_=$family->{$_}\n" for @$a;
1086 print "*$_=$family->{$_}\n" for @$c; 1107 print "*$_=$family->{$_}\n" for @$c;
1087 print "-$_=$family->{$_}\n" for @$d; 1108 print "-$_=$family->{$_}\n" for @$d;
1138filter messages without dequeuing them. 1159filter messages without dequeuing them.
1139 1160
1140This is not a philosophical difference, but simply stems from AnyEvent::MP 1161This is not a philosophical difference, but simply stems from AnyEvent::MP
1141being event-based, while Erlang is process-based. 1162being event-based, while Erlang is process-based.
1142 1163
1143You 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
1144top of AEMP and Coro threads. 1165top of AEMP and Coro threads.
1145 1166
1146=item * Erlang sends are synchronous, AEMP sends are asynchronous. 1167=item * Erlang sends are synchronous, AEMP sends are asynchronous.
1147 1168
1148Sending messages in Erlang is synchronous and blocks the process until 1169Sending messages in Erlang is synchronous and blocks the process until
1149a 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
1150need a queue that can overflow). AEMP sends return immediately, connection 1171need a queue that can overflow). AEMP sends return immediately, connection
1151establishment is handled in the background. 1172establishment is handled in the background.
1152 1173
1153=item * Erlang suffers from silent message loss, AEMP does not. 1174=item * Erlang suffers from silent message loss, AEMP does not.
1154 1175
1254 1275
1255=back 1276=back
1256 1277
1257=head1 PORTING FROM AnyEvent::MP VERSION 1.X 1278=head1 PORTING FROM AnyEvent::MP VERSION 1.X
1258 1279
1259AEMP version 2 has three major incompatible changes compared to version 1: 1280AEMP version 2 has a few major incompatible changes compared to version 1:
1260 1281
1261=over 4 1282=over 4
1262 1283
1263=item AnyEvent::MP::Global no longer has group management functions. 1284=item AnyEvent::MP::Global no longer has group management functions.
1264 1285
1286At least not officially - the grp_* functions are still exported and might
1287work, but they will be removed in some later release.
1288
1265AnyEvent::MP now comes with a distributed database that is more 1289AnyEvent::MP now comes with a distributed database that is more
1266powerful. It's database families map closely to ports, but the API has 1290powerful. Its database families map closely to port groups, but the API
1267minor differences: 1291has changed (the functions are also now exported by AnyEvent::MP). Here is
1292a rough porting guide:
1268 1293
1269 grp_reg $group, $port # old 1294 grp_reg $group, $port # old
1270 db_reg $group, $port # new 1295 db_reg $group, $port # new
1271 1296
1272 $list = grp_get $group # old 1297 $list = grp_get $group # old
1273 db_keys $group, sub { my $list = shift } # new 1298 db_keys $group, sub { my $list = shift } # new
1274 1299
1275 grp_mon $group, $cb->(\@ports, $add, $del) # old 1300 grp_mon $group, $cb->(\@ports, $add, $del) # old
1276 db_mon $group, $cb->(\%ports, $add, $change, $del) # new 1301 db_mon $group, $cb->(\%ports, $add, $change, $del) # new
1277 1302
1278C<grp_reg> is a no-brainer (just replace by C<db_reg>), but C<grp_get> 1303C<grp_reg> is a no-brainer (just replace by C<db_reg>), but C<grp_get> is
1279is no longer instant, because the local node might not have a copy of 1304no longer instant, because the local node might not have a copy of the
1280the group. This can be partially remedied by using C<db_mon> to keep an 1305group. You can either modify your code to allow for a callback, or use
1281updated copy of the group: 1306C<db_mon> to keep an updated copy of the group:
1282 1307
1283 my $local_group_copy; 1308 my $local_group_copy;
1284 db_mon $group => sub { $local_group_copy = shift }; 1309 db_mon $group => sub { $local_group_copy = $_[0] };
1285 1310
1286 # no keys %$local_group_copy always returns the most up-to-date 1311 # now "keys %$local_group_copy" always returns the most up-to-date
1287 # list of ports in the group. 1312 # list of ports in the group.
1288 1313
1289C<grp_mon> can almost be replaced by C<db_mon>: 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:
1290 1317
1291 db_mon $group => sub { 1318 db_mon $group => sub {
1292 my ($ports, $add, $chg, $lde) = @_; 1319 my ($ports, $add, $chg, $lde) = @_;
1293 $ports = [keys %$ports]; 1320 $ports = [keys %$ports];
1294 1321
1311of its seed nodes. If that seed node isn't a global node yet, it will 1338of its seed nodes. If that seed node isn't a global node yet, it will
1312automatically be upgraded to a global node. 1339automatically be upgraded to a global node.
1313 1340
1314So in many cases, nothing needs to be changed - one just has to make sure 1341So in many cases, nothing needs to be changed - one just has to make sure
1315that all seed nodes are meshed together with the other seed nodes (as with 1342that all seed nodes are meshed together with the other seed nodes (as with
1316AEMP 1.x), and other nodes specify them as seed nodes. 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.
1317 1346
1318Not opening a connection to every other node is usually an advantage, 1347Not opening a connection to every other node is usually an advantage,
1319except when you need the lower latency of an already established 1348except when you need the lower latency of an already established
1320connection. To ensure a node establishes a connection to another node, 1349connection. To ensure a node establishes a connection to another node,
1321you can monitor the node port (C<mon $node, ...>), which will attempt to 1350you can monitor the node port (C<mon $node, ...>), which will attempt to
1322create the connection (and notify you when the connection fails). 1351create the connection (and notify you when the connection fails).
1323 1352
1324=item Listener-less nodes (nodes without binds) are gone. 1353=item Listener-less nodes (nodes without binds) are gone.
1325 1354
1326And are not coming back, at least not in their old form. If no C<binds> 1355And are not coming back, at least not in their old form. If no C<binds>
1327are specified for a node, AnyEvent::MP now assumes a default of C<*:*>. 1356are specified for a node, AnyEvent::MP assumes a default of C<*:*>.
1328 1357
1329There are vague plans to implement some form of routing domains, which 1358There are vague plans to implement some form of routing domains, which
1330might or might not bring back listener-less nodes, but don't count on it. 1359might or might not bring back listener-less nodes, but don't count on it.
1331 1360
1332The fact that most connections are now optional somewhat mitigates this, 1361The fact that most connections are now optional somewhat mitigates this,
1341 1370
1342Every module now documents what kinds of messages it generates, with 1371Every module now documents what kinds of messages it generates, with
1343AnyEvent::MP acting as a catch all. 1372AnyEvent::MP acting as a catch all.
1344 1373
1345On the positive side, this means that instead of setting 1374On the positive side, this means that instead of setting
1346C<PERL_ANYEVENT_MP_WARNLEVEL>, you can get away by setting C<AE_VERBOSE>, 1375C<PERL_ANYEVENT_MP_WARNLEVEL>, you can get away by setting C<AE_VERBOSE> -
1347much less to type. 1376much less to type.
1348 1377
1349=back 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.
1350 1385
1351=head1 SEE ALSO 1386=head1 SEE ALSO
1352 1387
1353L<AnyEvent::MP::Intro> - a gentle introduction. 1388L<AnyEvent::MP::Intro> - a gentle introduction.
1354 1389

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines