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.48 by root, Thu Aug 13 02:59:42 2009 UTC vs.
Revision 1.50 by root, Fri Aug 14 14:01:05 2009 UTC

22 # sending messages 22 # sending messages
23 snd $port, type => data...; 23 snd $port, type => data...;
24 snd $port, @msg; 24 snd $port, @msg;
25 snd @msg_with_first_element_being_a_port; 25 snd @msg_with_first_element_being_a_port;
26 26
27 # creating/using miniports 27 # creating/using ports, the simple way
28 my $miniport = port { my @msg = @_; 0 }; 28 my $somple_port = port { my @msg = @_; 0 };
29 29
30 # creating/using full ports 30 # creating/using ports, type matching
31 my $port = port; 31 my $port = port;
32 rcv $port, smartmatch => $cb->(@msg);
33 rcv $port, ping => sub { snd $_[0], "pong"; 0 }; 32 rcv $port, ping => sub { snd $_[0], "pong"; 0 };
34 rcv $port, pong => sub { warn "pong received\n"; 0 }; 33 rcv $port, pong => sub { warn "pong received\n"; 0 };
35
36 # more, smarter, matches (_any_ is exported by this module)
37 rcv $port, [child_died => $pid] => sub { ...
38 rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3
39 34
40 # create a port on another node 35 # create a port on another node
41 my $port = spawn $node, $initfunc, @initdata; 36 my $port = spawn $node, $initfunc, @initdata;
42 37
43 # monitoring 38 # monitoring
168it should know the noderefs of some other nodes in the network. 163it should know the noderefs of some other nodes in the network.
169 164
170This function initialises a node - it must be called exactly once (or 165This function initialises a node - it must be called exactly once (or
171never) before calling other AnyEvent::MP functions. 166never) before calling other AnyEvent::MP functions.
172 167
173All arguments are noderefs, which can be either resolved or unresolved. 168All arguments (optionally except for the first) are noderefs, which can be
169either resolved or unresolved.
170
171The first argument will be looked up in the configuration database first
172(if it is C<undef> then the current nodename will be used instead) to find
173the relevant configuration profile (see L<aemp>). If none is found then
174the default configuration is used. The configuration supplies additional
175seed/master nodes and can override the actual noderef.
174 176
175There are two types of networked nodes, public nodes and slave nodes: 177There are two types of networked nodes, public nodes and slave nodes:
176 178
177=over 4 179=over 4
178 180
179=item public nodes 181=item public nodes
180 182
181For public nodes, C<$noderef> must either be a (possibly unresolved) 183For public nodes, C<$noderef> (supplied either directly to
182noderef, in which case it will be resolved, or C<undef> (or missing), in 184C<initialise_node> or indirectly via a profile or the nodename) must be a
183which case the noderef will be guessed. 185noderef (possibly unresolved, in which case it will be resolved).
184 186
185Afterwards, the node will bind itself on all endpoints and try to connect 187After resolving, the node will bind itself on all endpoints and try to
186to all additional C<$seednodes> that are specified. Seednodes are optional 188connect to all additional C<$seednodes> that are specified. Seednodes are
187and can be used to quickly bootstrap the node into an existing network. 189optional and can be used to quickly bootstrap the node into an existing
190network.
188 191
189=item slave nodes 192=item slave nodes
190 193
191When the C<$noderef> is the special string C<slave/>, then the node will 194When the C<$noderef> (either as given or overriden by the config file)
195is the special string C<slave/>, then the node will become a slave
192become a slave node. Slave nodes cannot be contacted from outside and will 196node. Slave nodes cannot be contacted from outside and will route most of
193route most of their traffic to the master node that they attach to. 197their traffic to the master node that they attach to.
194 198
195At least one additional noderef is required: The node will try to connect 199At least one additional noderef is required (either by specifying it
196to all of them and will become a slave attached to the first node it can 200directly or because it is part of the configuration profile): The node
197successfully connect to. 201will try to connect to all of them and will become a slave attached to the
202first node it can successfully connect to.
198 203
199=back 204=back
200 205
201This function will block until all nodes have been resolved and, for slave 206This function will block until all nodes have been resolved and, for slave
202nodes, until it has successfully established a connection to a master 207nodes, until it has successfully established a connection to a master
203server. 208server.
204 209
205Example: become a public node listening on the default node. 210Example: become a public node listening on the guessed noderef, or the one
211specified via C<aemp> for the current node. This should be the most common
212form of invocation for "daemon"-type nodes.
206 213
207 initialise_node; 214 initialise_node;
215
216Example: become a slave node to any of the the seednodes specified via
217C<aemp>. This form is often used for commandline clients.
218
219 initialise_node "slave/";
220
221Example: become a slave node to any of the specified master servers. This
222form is also often used for commandline clients.
223
224 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
208 225
209Example: become a public node, and try to contact some well-known master 226Example: become a public node, and try to contact some well-known master
210servers to become part of the network. 227servers to become part of the network.
211 228
212 initialise_node undef, "master1", "master2"; 229 initialise_node undef, "master1", "master2";
215 232
216 initialise_node 4041; 233 initialise_node 4041;
217 234
218Example: become a public node, only visible on localhost port 4044. 235Example: become a public node, only visible on localhost port 4044.
219 236
220 initialise_node "locahost:4044"; 237 initialise_node "localhost:4044";
221
222Example: become a slave node to any of the specified master servers.
223
224 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
225 238
226=item $cv = resolve_node $noderef 239=item $cv = resolve_node $noderef
227 240
228Takes an unresolved node reference that may contain hostnames and 241Takes an unresolved node reference that may contain hostnames and
229abbreviated IDs, resolves all of them and returns a resolved node 242abbreviated IDs, resolves all of them and returns a resolved node
285that Storable can serialise and deserialise is allowed, and for the local 298that Storable can serialise and deserialise is allowed, and for the local
286node, anything can be passed. 299node, anything can be passed.
287 300
288=item $local_port = port 301=item $local_port = port
289 302
290Create a new local port object that can be used either as a pattern 303Create a new local port object and returns its port ID. Initially it has
291matching port ("full port") or a single-callback port ("miniport"), 304no callbacks set and will throw an error when it receives messages.
292depending on how C<rcv> callbacks are bound to the object.
293 305
294=item $port = port { my @msg = @_; $finished } 306=item $local_port = port { my @msg = @_ }
295 307
296Creates a "miniport", that is, a very lightweight port without any pattern 308Creates a new local port, and returns its ID. Semantically the same as
297matching behind it, and returns its ID. Semantically the same as creating
298a port and calling C<rcv $port, $callback> on it. 309creating a port and calling C<rcv $port, $callback> on it.
299 310
300The block will be called for every message received on the port. When the 311The block will be called for every message received on the port, with the
301callback returns a true value its job is considered "done" and the port 312global variable C<$SELF> set to the port ID. Runtime errors will cause the
302will be destroyed. Otherwise it will stay alive. 313port to be C<kil>ed. The message will be passed as-is, no extra argument
314(i.e. no port ID) will be passed to the callback.
303 315
304The message will be passed as-is, no extra argument (i.e. no port id) will 316If you want to stop/destroy the port, simply C<kil> it:
305be passed to the callback.
306 317
307If you need the local port id in the callback, this works nicely: 318 my $port = port {
308 319 my @msg = @_;
309 my $port; $port = port { 320 ...
310 snd $otherport, reply => $port; 321 kil $SELF;
311 }; 322 };
312 323
313=cut 324=cut
314 325
315sub rcv($@); 326sub rcv($@);
327
328sub _kilme {
329 die "received message on port without callback";
330}
316 331
317sub port(;&) { 332sub port(;&) {
318 my $id = "$UNIQ." . $ID++; 333 my $id = "$UNIQ." . $ID++;
319 my $port = "$NODE#$id"; 334 my $port = "$NODE#$id";
320 335
321 if (@_) { 336 rcv $port, shift || \&_kilme;
322 rcv $port, shift;
323 } else {
324 $PORT{$id} = sub { }; # nop
325 }
326 337
327 $port 338 $port
328} 339}
329 340
330=item reg $port, $name
331
332=item reg $name
333
334Registers the given port (or C<$SELF><<< if missing) under the name
335C<$name>. If the name already exists it is replaced.
336
337A port can only be registered under one well known name.
338
339A port automatically becomes unregistered when it is killed.
340
341=cut
342
343sub reg(@) {
344 my $port = @_ > 1 ? shift : $SELF || Carp::croak 'reg: called with one argument only, but $SELF not set,';
345
346 $REG{$_[0]} = $port;
347}
348
349=item rcv $port, $callback->(@msg) 341=item rcv $local_port, $callback->(@msg)
350 342
351Replaces the callback on the specified miniport (after converting it to 343Replaces the default callback on the specified port. There is no way to
352one if required). 344remove the default callback: use C<sub { }> to disable it, or better
353 345C<kil> the port when it is no longer needed.
354=item rcv $port, tagstring => $callback->(@msg), ...
355
356=item rcv $port, $smartmatch => $callback->(@msg), ...
357
358=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
359
360Register callbacks to be called on matching messages on the given full
361port (after converting it to one if required) and return the port.
362
363The callback has to return a true value when its work is done, after
364which is will be removed, or a false value in which case it will stay
365registered.
366 346
367The global C<$SELF> (exported by this module) contains C<$port> while 347The global C<$SELF> (exported by this module) contains C<$port> while
368executing the callback. 348executing the callback. Runtime errors during callback execution will
349result in the port being C<kil>ed.
369 350
370Runtime errors during callback execution will result in the port being 351The default callback received all messages not matched by a more specific
371C<kil>ed. 352C<tag> match.
372 353
373If the match is an array reference, then it will be matched against the 354=item rcv $local_port, tag => $callback->(@msg_without_tag), ...
374first elements of the message, otherwise only the first element is being
375matched.
376 355
377Any element in the match that is specified as C<_any_> (a function 356Register callbacks to be called on messages starting with the given tag on
378exported by this module) matches any single element of the message. 357the given port (and return the port), or unregister it (when C<$callback>
358is C<$undef>).
379 359
380While not required, it is highly recommended that the first matching 360The original message will be passed to the callback, after the first
381element is a string identifying the message. The one-string-only match is 361element (the tag) has been removed. The callback will use the same
382also the most efficient match (by far). 362environment as the default callback (see above).
383 363
384Example: create a port and bind receivers on it in one go. 364Example: create a port and bind receivers on it in one go.
385 365
386 my $port = rcv port, 366 my $port = rcv port,
387 msg1 => sub { ...; 0 }, 367 msg1 => sub { ... },
388 msg2 => sub { ...; 0 }, 368 msg2 => sub { ... },
389 ; 369 ;
390 370
391Example: create a port, bind receivers and send it in a message elsewhere 371Example: create a port, bind receivers and send it in a message elsewhere
392in one go: 372in one go:
393 373
394 snd $otherport, reply => 374 snd $otherport, reply =>
395 rcv port, 375 rcv port,
396 msg1 => sub { ...; 0 }, 376 msg1 => sub { ... },
397 ... 377 ...
398 ; 378 ;
399 379
400=cut 380=cut
401 381
404 my ($noderef, $portid) = split /#/, $port, 2; 384 my ($noderef, $portid) = split /#/, $port, 2;
405 385
406 ($NODE{$noderef} || add_node $noderef) == $NODE{""} 386 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
407 or Carp::croak "$port: rcv can only be called on local ports, caught"; 387 or Carp::croak "$port: rcv can only be called on local ports, caught";
408 388
409 if (@_ == 1) { 389 while (@_) {
390 if (ref $_[0]) {
391 if (my $self = $PORT_DATA{$portid}) {
392 "AnyEvent::MP::Port" eq ref $self
393 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
394
395 $self->[2] = shift;
396 } else {
410 my $cb = shift; 397 my $cb = shift;
411 delete $PORT_DATA{$portid};
412 $PORT{$portid} = sub { 398 $PORT{$portid} = sub {
413 local $SELF = $port; 399 local $SELF = $port;
414 eval { 400 eval { &$cb }; _self_die if $@;
415 &$cb 401 };
416 and kil $port;
417 }; 402 }
418 _self_die if $@; 403 } elsif (defined $_[0]) {
419 };
420 } else {
421 my $self = $PORT_DATA{$portid} ||= do { 404 my $self = $PORT_DATA{$portid} ||= do {
422 my $self = bless { 405 my $self = bless [$PORT{$port} || sub { }, { }, $port], "AnyEvent::MP::Port";
423 id => $port,
424 }, "AnyEvent::MP::Port";
425 406
426 $PORT{$portid} = sub { 407 $PORT{$portid} = sub {
427 local $SELF = $port; 408 local $SELF = $port;
428 409
429 eval {
430 for (@{ $self->{rc0}{$_[0]} }) { 410 if (my $cb = $self->[1]{$_[0]}) {
431 $_ && &{$_->[0]} 411 shift;
432 && undef $_; 412 eval { &$cb }; _self_die if $@;
433 } 413 } else {
434
435 for (@{ $self->{rcv}{$_[0]} }) {
436 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
437 && &{$_->[0]} 414 &{ $self->[0] };
438 && undef $_;
439 }
440
441 for (@{ $self->{any} }) {
442 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
443 && &{$_->[0]}
444 && undef $_;
445 } 415 }
446 }; 416 };
447 _self_die if $@; 417
418 $self
448 }; 419 };
449 420
450 $self
451 };
452
453 "AnyEvent::MP::Port" eq ref $self 421 "AnyEvent::MP::Port" eq ref $self
454 or Carp::croak "$port: rcv can only be called on message matching ports, caught"; 422 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
455 423
456 while (@_) {
457 my ($match, $cb) = splice @_, 0, 2; 424 my ($tag, $cb) = splice @_, 0, 2;
458 425
459 if (!ref $match) { 426 if (defined $cb) {
460 push @{ $self->{rc0}{$match} }, [$cb]; 427 $self->[1]{$tag} = $cb;
461 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
462 my ($type, @match) = @$match;
463 @match
464 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
465 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
466 } else { 428 } else {
467 push @{ $self->{any} }, [$cb, $match]; 429 delete $self->[1]{$tag};
468 } 430 }
469 } 431 }
470 } 432 }
471 433
472 $port 434 $port

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines