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.31 by root, Wed Aug 5 19:55:58 2009 UTC vs.
Revision 1.34 by root, Wed Aug 5 23:50:46 2009 UTC

… …
123The C<NODE> function returns, and the C<$NODE> variable contains 123The C<NODE> function returns, and the C<$NODE> variable contains
124the noderef of the local node. The value is initialised by a call 124the noderef of the local node. The value is initialised by a call
125to C<become_public> or C<become_slave>, after which all local port 125to C<become_public> or C<become_slave>, after which all local port
126identifiers become invalid. 126identifiers become invalid.
127 127
128=item $noderef = node_of $portid 128=item $noderef = node_of $port
129 129
130Extracts and returns the noderef from a portid or a noderef. 130Extracts and returns the noderef from a portid or a noderef.
131
132=item initialise_node $noderef, $seednode, $seednode...
133
134=item initialise_node "slave/", $master, $master...
135
136Before a node can talk to other nodes on the network it has to initialise
137itself - the minimum a node needs to know is it's own name, and optionally
138it should know the noderefs of some other nodes in the network.
139
140This function initialises a node - it must be called exactly once (or
141never) before calling other AnyEvent::MP functions.
142
143All arguments are noderefs, which can be either resolved or unresolved.
144
145There are two types of networked nodes, public nodes and slave nodes:
146
147=over 4
148
149=item public nodes
150
151For public nodes, C<$noderef> must either be a (possibly unresolved)
152noderef, in which case it will be resolved, or C<undef> (or missing), in
153which case the noderef will be guessed.
154
155Afterwards, the node will bind itself on all endpoints and try to connect
156to all additional C<$seednodes> that are specified. Seednodes are optional
157and can be used to quickly bootstrap the node into an existing network.
158
159=item slave nodes
160
161When the C<$noderef> is the special string C<slave/>, then the node will
162become a slave node. Slave nodes cannot be contacted from outside and will
163route most of their traffic to the master node that they attach to.
164
165At least one additional noderef is required: The node will try to connect
166to all of them and will become a slave attached to the first node it can
167successfully connect to.
168
169=back
170
171This function will block until all nodes have been resolved and, for slave
172nodes, until it has successfully established a connection to a master
173server.
174
175Example: become a public node listening on the default node.
176
177 initialise_node;
178
179Example: become a public node, and try to contact some well-known master
180servers to become part of the network.
181
182 initialise_node undef, "master1", "master2";
183
184Example: become a public node listening on port C<4041>.
185
186 initialise_node 4041;
187
188Example: become a public node, only visible on localhost port 4044.
189
190 initialise_node "locahost:4044";
191
192Example: become a slave node to any of the specified master servers.
193
194 initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
131 195
132=item $cv = resolve_node $noderef 196=item $cv = resolve_node $noderef
133 197
134Takes an unresolved node reference that may contain hostnames and 198Takes an unresolved node reference that may contain hostnames and
135abbreviated IDs, resolves all of them and returns a resolved node 199abbreviated IDs, resolves all of them and returns a resolved node
… …
167 231
168Due to some quirks in how perl exports variables, it is impossible to 232Due to some quirks in how perl exports variables, it is impossible to
169just export C<$SELF>, all the symbols called C<SELF> are exported by this 233just export C<$SELF>, all the symbols called C<SELF> are exported by this
170module, but only C<$SELF> is currently used. 234module, but only C<$SELF> is currently used.
171 235
172=item snd $portid, type => @data 236=item snd $port, type => @data
173 237
174=item snd $portid, @msg 238=item snd $port, @msg
175 239
176Send the given message to the given port ID, which can identify either 240Send the given message to the given port ID, which can identify either
177a local or a remote port, and can be either a string or soemthignt hat 241a local or a remote port, and can be either a string or soemthignt hat
178stringifies a sa port ID (such as a port object :). 242stringifies a sa port ID (such as a port object :).
179 243
… …
189JSON is used, then only strings, numbers and arrays and hashes consisting 253JSON is used, then only strings, numbers and arrays and hashes consisting
190of those are allowed (no objects). When Storable is used, then anything 254of those are allowed (no objects). When Storable is used, then anything
191that Storable can serialise and deserialise is allowed, and for the local 255that Storable can serialise and deserialise is allowed, and for the local
192node, anything can be passed. 256node, anything can be passed.
193 257
194=item kil $portid[, @reason] 258=item $local_port = port
195 259
196Kill the specified port with the given C<@reason>. 260Create a new local port object that can be used either as a pattern
261matching port ("full port") or a single-callback port ("miniport"),
262depending on how C<rcv> callbacks are bound to the object.
197 263
198If no C<@reason> is specified, then the port is killed "normally" (linked 264=item $port = port { my @msg = @_; $finished }
199ports will not be kileld, or even notified).
200 265
201Otherwise, linked ports get killed with the same reason (second form of 266Creates a "miniport", that is, a very lightweight port without any pattern
202C<mon>, see below). 267matching behind it, and returns its ID. Semantically the same as creating
268a port and calling C<rcv $port, $callback> on it.
203 269
204Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks 270The block will be called for every message received on the port. When the
205will be reported as reason C<< die => $@ >>. 271callback returns a true value its job is considered "done" and the port
272will be destroyed. Otherwise it will stay alive.
206 273
207Transport/communication errors are reported as C<< transport_error => 274The message will be passed as-is, no extra argument (i.e. no port id) will
208$message >>. 275be passed to the callback.
209 276
277If you need the local port id in the callback, this works nicely:
278
279 my $port; $port = port {
280 snd $otherport, reply => $port;
281 };
282
283=cut
284
285sub rcv($@);
286
287sub port(;&) {
288 my $id = "$UNIQ." . $ID++;
289 my $port = "$NODE#$id";
290
291 if (@_) {
292 rcv $port, shift;
293 } else {
294 $PORT{$id} = sub { }; # nop
295 }
296
297 $port
298}
299
300=item reg $port, $name
301
302Registers the given port under the name C<$name>. If the name already
303exists it is replaced.
304
305A port can only be registered under one well known name.
306
307A port automatically becomes unregistered when it is killed.
308
309=cut
310
311sub reg(@) {
312 my ($port, $name) = @_;
313
314 $REG{$name} = $port;
315}
316
317=item rcv $port, $callback->(@msg)
318
319Replaces the callback on the specified miniport (after converting it to
320one if required).
321
322=item rcv $port, tagstring => $callback->(@msg), ...
323
324=item rcv $port, $smartmatch => $callback->(@msg), ...
325
326=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
327
328Register callbacks to be called on matching messages on the given full
329port (after converting it to one if required).
330
331The callback has to return a true value when its work is done, after
332which is will be removed, or a false value in which case it will stay
333registered.
334
335The global C<$SELF> (exported by this module) contains C<$port> while
336executing the callback.
337
338Runtime errors wdurign callback execution will result in the port being
339C<kil>ed.
340
341If the match is an array reference, then it will be matched against the
342first elements of the message, otherwise only the first element is being
343matched.
344
345Any element in the match that is specified as C<_any_> (a function
346exported by this module) matches any single element of the message.
347
348While not required, it is highly recommended that the first matching
349element is a string identifying the message. The one-string-only match is
350also the most efficient match (by far).
351
352=cut
353
354sub rcv($@) {
355 my $port = shift;
356 my ($noderef, $portid) = split /#/, $port, 2;
357
358 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
359 or Carp::croak "$port: rcv can only be called on local ports, caught";
360
361 if (@_ == 1) {
362 my $cb = shift;
363 delete $PORT_DATA{$portid};
364 $PORT{$portid} = sub {
365 local $SELF = $port;
366 eval {
367 &$cb
368 and kil $port;
369 };
370 _self_die if $@;
371 };
372 } else {
373 my $self = $PORT_DATA{$portid} ||= do {
374 my $self = bless {
375 id => $port,
376 }, "AnyEvent::MP::Port";
377
378 $PORT{$portid} = sub {
379 local $SELF = $port;
380
381 eval {
382 for (@{ $self->{rc0}{$_[0]} }) {
383 $_ && &{$_->[0]}
384 && undef $_;
385 }
386
387 for (@{ $self->{rcv}{$_[0]} }) {
388 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
389 && &{$_->[0]}
390 && undef $_;
391 }
392
393 for (@{ $self->{any} }) {
394 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
395 && &{$_->[0]}
396 && undef $_;
397 }
398 };
399 _self_die if $@;
400 };
401
402 $self
403 };
404
405 "AnyEvent::MP::Port" eq ref $self
406 or Carp::croak "$port: rcv can only be called on message matching ports, caught";
407
408 while (@_) {
409 my ($match, $cb) = splice @_, 0, 2;
410
411 if (!ref $match) {
412 push @{ $self->{rc0}{$match} }, [$cb];
413 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
414 my ($type, @match) = @$match;
415 @match
416 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
417 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
418 } else {
419 push @{ $self->{any} }, [$cb, $match];
420 }
421 }
422 }
423
424 $port
425}
426
427=item $closure = psub { BLOCK }
428
429Remembers C<$SELF> and creates a closure out of the BLOCK. When the
430closure is executed, sets up the environment in the same way as in C<rcv>
431callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
432
433This is useful when you register callbacks from C<rcv> callbacks:
434
435 rcv delayed_reply => sub {
436 my ($delay, @reply) = @_;
437 my $timer = AE::timer $delay, 0, psub {
438 snd @reply, $SELF;
439 };
440 };
441
442=cut
443
444sub psub(&) {
445 my $cb = shift;
446
447 my $port = $SELF
448 or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
449
450 sub {
451 local $SELF = $port;
452
453 if (wantarray) {
454 my @res = eval { &$cb };
455 _self_die if $@;
456 @res
457 } else {
458 my $res = eval { &$cb };
459 _self_die if $@;
460 $res
461 }
462 }
463}
464
210=item $guard = mon $portid, $cb->(@reason) 465=item $guard = mon $port, $cb->(@reason)
211 466
212=item $guard = mon $portid, $otherport 467=item $guard = mon $port, $otherport
213 468
214=item $guard = mon $portid, $otherport, @msg 469=item $guard = mon $port, $otherport, @msg
215 470
216Monitor the given port and do something when the port is killed. 471Monitor the given port and do something when the port is killed.
217 472
218In the first form, the callback is simply called with any number 473In the first form, the callback is simply called with any number
219of C<@reason> elements (no @reason means that the port was deleted 474of C<@reason> elements (no @reason means that the port was deleted
… …
297 mon $port2, $port1; 552 mon $port2, $port1;
298 553
299It means that if either one is killed abnormally, the other one gets 554It means that if either one is killed abnormally, the other one gets
300killed as well. 555killed as well.
301 556
302=item $local_port = port 557=item kil $port[, @reason]
303 558
304Create a new local port object that can be used either as a pattern 559Kill the specified port with the given C<@reason>.
305matching port ("full port") or a single-callback port ("miniport"),
306depending on how C<rcv> callbacks are bound to the object.
307 560
308=item $portid = port { my @msg = @_; $finished } 561If no C<@reason> is specified, then the port is killed "normally" (linked
562ports will not be kileld, or even notified).
309 563
310Creates a "mini port", that is, a very lightweight port without any 564Otherwise, linked ports get killed with the same reason (second form of
311pattern matching behind it, and returns its ID. 565C<mon>, see below).
312 566
313The block will be called for every message received on the port. When the 567Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
314callback returns a true value its job is considered "done" and the port 568will be reported as reason C<< die => $@ >>.
315will be destroyed. Otherwise it will stay alive.
316 569
317The message will be passed as-is, no extra argument (i.e. no port id) will 570Transport/communication errors are reported as C<< transport_error =>
318be passed to the callback. 571$message >>.
319
320If you need the local port id in the callback, this works nicely:
321
322 my $port; $port = port {
323 snd $otherport, reply => $port;
324 };
325
326=cut
327
328sub port(;&) {
329 my $id = "$UNIQ." . $ID++;
330 my $port = "$NODE#$id";
331
332 if (@_) {
333 my $cb = shift;
334 $PORT{$id} = sub {
335 local $SELF = $port;
336 eval {
337 &$cb
338 and kil $id;
339 };
340 _self_die if $@;
341 };
342 } else {
343 my $self = bless {
344 id => "$NODE#$id",
345 }, "AnyEvent::MP::Port";
346
347 $PORT_DATA{$id} = $self;
348 $PORT{$id} = sub {
349 local $SELF = $port;
350
351 eval {
352 for (@{ $self->{rc0}{$_[0]} }) {
353 $_ && &{$_->[0]}
354 && undef $_;
355 }
356
357 for (@{ $self->{rcv}{$_[0]} }) {
358 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
359 && &{$_->[0]}
360 && undef $_;
361 }
362
363 for (@{ $self->{any} }) {
364 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
365 && &{$_->[0]}
366 && undef $_;
367 }
368 };
369 _self_die if $@;
370 };
371 }
372
373 $port
374}
375
376=item reg $portid, $name
377
378Registers the given port under the name C<$name>. If the name already
379exists it is replaced.
380
381A port can only be registered under one well known name.
382
383A port automatically becomes unregistered when it is killed.
384
385=cut
386
387sub reg(@) {
388 my ($portid, $name) = @_;
389
390 $REG{$name} = $portid;
391}
392
393=item rcv $portid, $callback->(@msg)
394
395Replaces the callback on the specified miniport (or newly created port
396object, see C<port>). Full ports are configured with the following calls:
397
398=item rcv $portid, tagstring => $callback->(@msg), ...
399
400=item rcv $portid, $smartmatch => $callback->(@msg), ...
401
402=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
403
404Register callbacks to be called on matching messages on the given port.
405
406The callback has to return a true value when its work is done, after
407which is will be removed, or a false value in which case it will stay
408registered.
409
410The global C<$SELF> (exported by this module) contains C<$portid> while
411executing the callback.
412
413Runtime errors wdurign callback execution will result in the port being
414C<kil>ed.
415
416If the match is an array reference, then it will be matched against the
417first elements of the message, otherwise only the first element is being
418matched.
419
420Any element in the match that is specified as C<_any_> (a function
421exported by this module) matches any single element of the message.
422
423While not required, it is highly recommended that the first matching
424element is a string identifying the message. The one-string-only match is
425also the most efficient match (by far).
426
427=cut
428
429sub rcv($@) {
430 my $portid = shift;
431 my ($noderef, $port) = split /#/, $port, 2;
432
433 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
434 or Carp::croak "$noderef#$port: rcv can only be called on local ports, caught";
435
436 my $self = $PORT_DATA{$port}
437 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
438
439 "AnyEvent::MP::Port" eq ref $self
440 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
441
442 while (@_) {
443 my ($match, $cb) = splice @_, 0, 2;
444
445 if (!ref $match) {
446 push @{ $self->{rc0}{$match} }, [$cb];
447 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
448 my ($type, @match) = @$match;
449 @match
450 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
451 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
452 } else {
453 push @{ $self->{any} }, [$cb, $match];
454 }
455 }
456
457 $portid
458}
459
460=item $closure = psub { BLOCK }
461
462Remembers C<$SELF> and creates a closure out of the BLOCK. When the
463closure is executed, sets up the environment in the same way as in C<rcv>
464callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
465
466This is useful when you register callbacks from C<rcv> callbacks:
467
468 rcv delayed_reply => sub {
469 my ($delay, @reply) = @_;
470 my $timer = AE::timer $delay, 0, psub {
471 snd @reply, $SELF;
472 };
473 };
474
475=cut
476
477sub psub(&) {
478 my $cb = shift;
479
480 my $port = $SELF
481 or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
482
483 sub {
484 local $SELF = $port;
485
486 if (wantarray) {
487 my @res = eval { &$cb };
488 _self_die if $@;
489 @res
490 } else {
491 my $res = eval { &$cb };
492 _self_die if $@;
493 $res
494 }
495 }
496}
497
498=back
499
500=head1 FUNCTIONS FOR NODES
501
502=over 4
503
504=item become_public $noderef
505
506Tells the node to become a public node, i.e. reachable from other nodes.
507
508The first argument is the (unresolved) node reference of the local node
509(if missing then the empty string is used).
510
511It is quite common to not specify anything, in which case the local node
512tries to listen on the default port, or to only specify a port number, in
513which case AnyEvent::MP tries to guess the local addresses.
514
515=cut
516 572
517=back 573=back
518 574
519=head1 NODE MESSAGES 575=head1 NODE MESSAGES
520 576

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines