ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/cvsroot/AnyEvent-MP/MP.pm
(Generate patch)

Comparing cvsroot/AnyEvent-MP/MP.pm (file contents):
Revision 1.142 by root, Fri Mar 23 13:44:01 2012 UTC vs.
Revision 1.153 by root, Sat Nov 2 01:30:49 2019 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 # 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
56
57 # distributed database - queries
58 db_family $family => $cb->(\%familyhash)
59 db_keys $family => $cb->(\@keys)
60 db_values $family => $cb->(\@values)
61
62 # distributed database - monitoring a family
63 db_mon $family => $cb->(\%familyhash, \@added, \@changed, \@deleted)
64
52=head1 DESCRIPTION 65=head1 DESCRIPTION
53 66
54This module (-family) implements a simple message passing framework. 67This module (-family) implements a simple message passing framework.
55 68
56Despite its simplicity, you can securely message other processes running 69Despite its simplicity, you can securely message other processes running
105each 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
106endpoints - binds. 119endpoints - binds.
107 120
108Currently, only standard C<ip:port> specifications can be used, which 121Currently, only standard C<ip:port> specifications can be used, which
109specify 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
110in listening mode thta accepts conenctions form other nodes. 123in listening mode that accepts connections from other nodes.
111 124
112=item seed nodes 125=item seed nodes
113 126
114When 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
115needs 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
116network. These other nodes are called "seed nodes". 129network. These other nodes are called "seed nodes".
117 130
118Seed nodes themselves are not special - they are seed nodes only because 131Seed nodes themselves are not special - they are seed nodes only because
119some 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
120node 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.
121 134
122In addition to discovering the network, seed nodes are also used to 135In addition to discovering the network, seed nodes are also used to
123maintain 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
124the 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
125the 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
126nodes for all nodes ensures that eventually the subnets get merged again. 139for all nodes ensures that eventually the subnets get merged again.
127 140
128Seed 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
129should always be available. They should also be relatively responsive - a 142should always be available. They should also be relatively responsive - a
130seed node that blocks for long periods will slow down everybody else. 143seed node that blocks for long periods will slow down everybody else.
131 144
155 168
156Any 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
157node 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
158make 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
159sense 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
160keep 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
161reduces overhead). 174reduces overhead).
162 175
163=back 176=back
164 177
165=head1 VARIABLES/FUNCTIONS 178=head1 VARIABLES/FUNCTIONS
170 183
171package AnyEvent::MP; 184package AnyEvent::MP;
172 185
173use AnyEvent::MP::Config (); 186use AnyEvent::MP::Config ();
174use AnyEvent::MP::Kernel; 187use AnyEvent::MP::Kernel;
175use 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);
176 199
177use common::sense; 200use common::sense;
178 201
179use Carp (); 202use Carp ();
180 203
181use AnyEvent (); 204use AnyEvent ();
182use Guard (); 205use Guard ();
183 206
184use base "Exporter"; 207use base "Exporter";
185 208
186our $VERSION = $AnyEvent::MP::Config::VERSION; 209our $VERSION = '2.02'; # also in MP/Config.pm
187 210
188our @EXPORT = qw( 211our @EXPORT = qw(
189 NODE $NODE *SELF node_of after
190 configure 212 configure
213
214 NODE $NODE
215 *SELF
216
217 node_of port_is_local
218
219 snd kil
191 snd rcv mon mon_guard kil psub peval spawn cal 220 port rcv mon mon_guard psub peval spawn cal
192 port
193 db_set db_del db_reg 221 db_set db_del db_reg
194 db_mon db_family db_keys db_values 222 db_mon db_family db_keys db_values
223
224 after
195); 225);
196 226
197our $SELF; 227our $SELF;
198 228
199sub _self_die() { 229sub _self_die() {
210 240
211=item $nodeid = node_of $port 241=item $nodeid = node_of $port
212 242
213Extracts and returns the node ID from a port ID or a node ID. 243Extracts and returns the node ID from a port ID or a node ID.
214 244
245=item $is_local = port_is_local $port
246
247Returns true iff the port is a local port.
248
215=item configure $profile, key => value... 249=item configure $profile, key => value...
216 250
217=item configure key => value... 251=item configure key => value...
218 252
219Before a node can talk to other nodes on the network (i.e. enter 253Before a node can talk to other nodes on the network (i.e. enter
230=over 4 264=over 4
231 265
232=item norc => $boolean (default false) 266=item norc => $boolean (default false)
233 267
234If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not> 268If true, then the rc file (e.g. F<~/.perl-anyevent-mp>) will I<not>
235be consulted - all configuraiton options must be specified in the 269be consulted - all configuration options must be specified in the
236C<configure> call. 270C<configure> call.
237 271
238=item force => $boolean (default false) 272=item force => $boolean (default false)
239 273
240IF true, then the values specified in the C<configure> will take 274IF true, then the values specified in the C<configure> will take
276 310
277=item step 2, bind listener sockets 311=item step 2, bind listener sockets
278 312
279The next step is to look up the binds in the profile, followed by binding 313The next step is to look up the binds in the profile, followed by binding
280aemp protocol listeners on all binds specified (it is possible and valid 314aemp protocol listeners on all binds specified (it is possible and valid
281to have no binds, meaning that the node cannot be contacted form the 315to have no binds, meaning that the node cannot be contacted from the
282outside. This means the node cannot talk to other nodes that also have no 316outside. This means the node cannot talk to other nodes that also have no
283binds, but it can still talk to all "normal" nodes). 317binds, but it can still talk to all "normal" nodes).
284 318
285If the profile does not specify a binds list, then a default of C<*> is 319If the profile does not specify a binds list, then a default of C<*> is
286used, meaning the node will bind on a dynamically-assigned port on every 320used, meaning the node will bind on a dynamically-assigned port on every
383=cut 417=cut
384 418
385sub rcv($@); 419sub rcv($@);
386 420
387my $KILME = sub { 421my $KILME = sub {
388 (my $tag = substr $_[0], 0, 30) =~ s/([\x20-\x7e])/./g; 422 (my $tag = substr $_[0], 0, 30) =~ s/([^\x20-\x7e])/./g;
389 kil $SELF, unhandled_message => "no callback found for message '$tag'"; 423 kil $SELF, unhandled_message => "no callback found for message '$tag'";
390}; 424};
391 425
392sub port(;&) { 426sub port(;&) {
393 my $id = $UNIQ . ++$ID; 427 my $id = $UNIQ . ++$ID;
451 485
452sub rcv($@) { 486sub rcv($@) {
453 my $port = shift; 487 my $port = shift;
454 my ($nodeid, $portid) = split /#/, $port, 2; 488 my ($nodeid, $portid) = split /#/, $port, 2;
455 489
456 $NODE{$nodeid} == $NODE{""} 490 $nodeid eq $NODE
457 or Carp::croak "$port: rcv can only be called on local ports, caught"; 491 or Carp::croak "$port: rcv can only be called on local ports, caught";
458 492
459 while (@_) { 493 while (@_) {
460 if (ref $_[0]) { 494 if (ref $_[0]) {
461 if (my $self = $PORT_DATA{$portid}) { 495 if (my $self = $PORT_DATA{$portid}) {
504 $port 538 $port
505} 539}
506 540
507=item peval $port, $coderef[, @args] 541=item peval $port, $coderef[, @args]
508 542
509Evaluates the given C<$codref> within the contetx of C<$port>, that is, 543Evaluates the given C<$codref> within the context of C<$port>, that is,
510when the code throews an exception the C<$port> will be killed. 544when the code throws an exception the C<$port> will be killed.
511 545
512Any remaining args will be passed to the callback. Any return values will 546Any remaining args will be passed to the callback. Any return values will
513be returned to the caller. 547be returned to the caller.
514 548
515This is useful when you temporarily want to execute code in the context of 549This is useful when you temporarily want to execute code in the context of
1020=item db_values $family => $cb->(\@values) 1054=item db_values $family => $cb->(\@values)
1021 1055
1022Same as C<db_family>, except it only queries the family I<values> and passes them 1056Same as C<db_family>, except it only queries the family I<values> and passes them
1023as array reference to the callback. 1057as array reference to the callback.
1024 1058
1025=item $guard = db_mon $family => $cb->($familyhash, \@added, \@changed, \@deleted) 1059=item $guard = db_mon $family => $cb->(\%familyhash, \@added, \@changed, \@deleted)
1026 1060
1027Creates a monitor on the given database family. Each time a key is set 1061Creates a monitor on the given database family. Each time a key is
1028or or is deleted the callback is called with a hash containing the 1062set or is deleted the callback is called with a hash containing the
1029database family and three lists of added, changed and deleted subkeys, 1063database family and three lists of added, changed and deleted subkeys,
1030respectively. If no keys have changed then the array reference might be 1064respectively. If no keys have changed then the array reference might be
1031C<undef> or even missing. 1065C<undef> or even missing.
1032 1066
1033If not called in void context, a guard object is returned that, when 1067If not called in void context, a guard object is returned that, when
1061 return unless %$family; 1095 return unless %$family;
1062 undef $guard; 1096 undef $guard;
1063 print "My::Module::workers now nonempty\n"; 1097 print "My::Module::workers now nonempty\n";
1064 }; 1098 };
1065 1099
1066Example: print all changes to the family "AnyRvent::Fantasy::Module". 1100Example: print all changes to the family "AnyEvent::Fantasy::Module".
1067 1101
1068 my $guard = db_mon AnyRvent::Fantasy::Module => sub { 1102 my $guard = db_mon AnyEvent::Fantasy::Module => sub {
1069 my ($family, $a, $c, $d) = @_; 1103 my ($family, $a, $c, $d) = @_;
1070 1104
1071 print "+$_=$family->{$_}\n" for @$a; 1105 print "+$_=$family->{$_}\n" for @$a;
1072 print "*$_=$family->{$_}\n" for @$c; 1106 print "*$_=$family->{$_}\n" for @$c;
1073 print "-$_=$family->{$_}\n" for @$d; 1107 print "-$_=$family->{$_}\n" for @$d;
1124filter messages without dequeuing them. 1158filter messages without dequeuing them.
1125 1159
1126This is not a philosophical difference, but simply stems from AnyEvent::MP 1160This is not a philosophical difference, but simply stems from AnyEvent::MP
1127being event-based, while Erlang is process-based. 1161being event-based, while Erlang is process-based.
1128 1162
1129You cna have a look at L<Coro::MP> for a more Erlang-like process model on 1163You can have a look at L<Coro::MP> for a more Erlang-like process model on
1130top of AEMP and Coro threads. 1164top of AEMP and Coro threads.
1131 1165
1132=item * Erlang sends are synchronous, AEMP sends are asynchronous. 1166=item * Erlang sends are synchronous, AEMP sends are asynchronous.
1133 1167
1134Sending messages in Erlang is synchronous and blocks the process until 1168Sending messages in Erlang is synchronous and blocks the process until
1135a conenction has been established and the message sent (and so does not 1169a connection has been established and the message sent (and so does not
1136need a queue that can overflow). AEMP sends return immediately, connection 1170need a queue that can overflow). AEMP sends return immediately, connection
1137establishment is handled in the background. 1171establishment is handled in the background.
1138 1172
1139=item * Erlang suffers from silent message loss, AEMP does not. 1173=item * Erlang suffers from silent message loss, AEMP does not.
1140 1174
1279C<grp_mon> can be replaced by C<db_mon> with minor changes - C<db_mon> 1313C<grp_mon> can be replaced by C<db_mon> with minor changes - C<db_mon>
1280passes a hash as first argument, and an extra C<$chg> argument that can be 1314passes a hash as first argument, and an extra C<$chg> argument that can be
1281ignored: 1315ignored:
1282 1316
1283 db_mon $group => sub { 1317 db_mon $group => sub {
1284 my ($ports, $add, $chg, $lde) = @_; 1318 my ($ports, $add, $chg, $del) = @_;
1285 $ports = [keys %$ports]; 1319 $ports = [keys %$ports];
1286 1320
1287 # now $ports, $add and $del are the same as 1321 # now $ports, $add and $del are the same as
1288 # were originally passed by grp_mon. 1322 # were originally passed by grp_mon.
1289 ... 1323 ...
1342 1376
1343=back 1377=back
1344 1378
1345=head1 LOGGING 1379=head1 LOGGING
1346 1380
1347AnyEvent::MP does not normally log anything by itself, but sinc eit is the 1381AnyEvent::MP does not normally log anything by itself, but since it is the
1348root of the contetx hierarchy for AnyEvent::MP modules, it will receive 1382root of the context hierarchy for AnyEvent::MP modules, it will receive
1349all log messages by submodules. 1383all log messages by submodules.
1350 1384
1351=head1 SEE ALSO 1385=head1 SEE ALSO
1352 1386
1353L<AnyEvent::MP::Intro> - a gentle introduction. 1387L<AnyEvent::MP::Intro> - a gentle introduction.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines