ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.pm
(Generate patch)

Comparing IO-AIO/AIO.pm (file contents):
Revision 1.57 by root, Sun Oct 22 01:28:31 2006 UTC vs.
Revision 1.63 by root, Mon Oct 23 00:34:36 2006 UTC

79 our $VERSION = '2.0'; 79 our $VERSION = '2.0';
80 80
81 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 81 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
82 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink 82 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
83 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move 83 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
84 aio_group); 84 aio_group aio_nop);
85 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs); 85 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs);
86 86
87 @IO::AIO::GRP::ISA = 'IO::AIO::REQ'; 87 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
88 88
89 require XSLoader; 89 require XSLoader;
182 print "read $_[0] bytes: <$buffer>\n"; 182 print "read $_[0] bytes: <$buffer>\n";
183 }; 183 };
184 184
185=item aio_move $srcpath, $dstpath, $callback->($status) 185=item aio_move $srcpath, $dstpath, $callback->($status)
186 186
187[EXPERIMENTAL due to internal aio_group use]
188
187Try to move the I<file> (directories not supported as either source or 189Try to move the I<file> (directories not supported as either source or
188destination) from C<$srcpath> to C<$dstpath> and call the callback with 190destination) from C<$srcpath> to C<$dstpath> and call the callback with
189the C<0> (error) or C<-1> ok. 191the C<0> (error) or C<-1> ok.
190 192
191This is a composite request that tries to rename(2) the file first. If 193This is a composite request that tries to rename(2) the file first. If
201=cut 203=cut
202 204
203sub aio_move($$$) { 205sub aio_move($$$) {
204 my ($src, $dst, $cb) = @_; 206 my ($src, $dst, $cb) = @_;
205 207
206 my $grp = aio_group; 208 my $grp = aio_group $cb;
207 209
208 add $grp aio_rename $src, $dst, sub { 210 add $grp aio_rename $src, $dst, sub {
209 if ($_[0] && $! == EXDEV) { 211 if ($_[0] && $! == EXDEV) {
210 add $grp aio_open $src, O_RDONLY, 0, sub { 212 add $grp aio_open $src, O_RDONLY, 0, sub {
211 if (my $src_fh = $_[0]) { 213 if (my $src_fh = $_[0]) {
221 chmod $stat[2] & 07777, $dst_fh; 223 chmod $stat[2] & 07777, $dst_fh;
222 chown $stat[4], $stat[5], $dst_fh; 224 chown $stat[4], $stat[5], $dst_fh;
223 close $dst_fh; 225 close $dst_fh;
224 226
225 add $grp aio_unlink $src, sub { 227 add $grp aio_unlink $src, sub {
226 $cb->($_[0]); 228 $grp->result ($_[0]);
227 }; 229 };
228 } else { 230 } else {
229 my $errno = $!; 231 my $errno = $!;
230 add $grp aio_unlink $dst, sub { 232 add $grp aio_unlink $dst, sub {
231 $! = $errno; 233 $! = $errno;
232 $cb->(-1); 234 $grp->result (-1);
233 }; 235 };
234 } 236 }
235 }; 237 };
236 } else { 238 } else {
237 $cb->(-1); 239 $grp->result (-1);
238 } 240 }
239 }, 241 },
240 242
241 } else { 243 } else {
242 $cb->(-1); 244 $grp->result (-1);
243 } 245 }
244 }; 246 };
245 } else { 247 } else {
246 $cb->($_[0]); 248 $grp->result ($_[0]);
247 } 249 }
248 }; 250 };
249 251
250 $grp 252 $grp
251} 253}
343The callback a single argument which is either C<undef> or an array-ref 345The callback a single argument which is either C<undef> or an array-ref
344with the filenames. 346with the filenames.
345 347
346=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 348=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
347 349
350[EXPERIMENTAL due to internal aio_group use]
351
348Scans a directory (similar to C<aio_readdir>) but additionally tries to 352Scans a directory (similar to C<aio_readdir>) but additionally tries to
349separate the entries of directory C<$path> into two sets of names, ones 353separate the entries of directory C<$path> into two sets of names, ones
350you can recurse into (directories or links to them), and ones you cannot 354you can recurse into (directories or links to them), and ones you cannot
351recurse into (everything else). 355recurse into (everything else).
352 356
353C<aio_scandir> is a composite request that consists of many sub 357C<aio_scandir> is a composite request that creates of many sub requests_
354requests. C<$maxreq> specifies the maximum number of outstanding aio 358C<$maxreq> specifies the maximum number of outstanding aio requests that
355requests that this function generates. If it is C<< <= 0 >>, then a 359this function generates. If it is C<< <= 0 >>, then a suitable default
356suitable default will be chosen (currently 8). 360will be chosen (currently 6).
357 361
358On error, the callback is called without arguments, otherwise it receives 362On error, the callback is called without arguments, otherwise it receives
359two array-refs with path-relative entry names. 363two array-refs with path-relative entry names.
360 364
361Example: 365Example:
398=cut 402=cut
399 403
400sub aio_scandir($$$) { 404sub aio_scandir($$$) {
401 my ($path, $maxreq, $cb) = @_; 405 my ($path, $maxreq, $cb) = @_;
402 406
403 my $grp = aio_group; 407 my $grp = aio_group $cb;
404 408
405 $maxreq = 8 if $maxreq <= 0; 409 $maxreq = 6 if $maxreq <= 0;
406 410
407 # stat once 411 # stat once
408 add $grp aio_stat $path, sub { 412 add $grp aio_stat $path, sub {
409 return $cb->() if $_[0]; 413 return $grp->result () if $_[0];
410 my $now = time; 414 my $now = time;
411 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 415 my $hash1 = join ":", (stat _)[0,1,3,7,9];
412 416
413 # read the directory entries 417 # read the directory entries
414 add $grp aio_readdir $path, sub { 418 add $grp aio_readdir $path, sub {
415 my $entries = shift 419 my $entries = shift
416 or return $cb->(); 420 or return $grp->result ();
417 421
418 # stat the dir another time 422 # stat the dir another time
419 add $grp aio_stat $path, sub { 423 add $grp aio_stat $path, sub {
420 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 424 my $hash2 = join ":", (stat _)[0,1,3,7,9];
421 425
426 $ndirs = -1; 430 $ndirs = -1;
427 } else { 431 } else {
428 # if nlink == 2, we are finished 432 # if nlink == 2, we are finished
429 # on non-posix-fs's, we rely on nlink < 2 433 # on non-posix-fs's, we rely on nlink < 2
430 $ndirs = (stat _)[3] - 2 434 $ndirs = (stat _)[3] - 2
431 or return $cb->([], $entries); 435 or return $grp->result ([], $entries);
432 } 436 }
433 437
434 # sort into likely dirs and likely nondirs 438 # sort into likely dirs and likely nondirs
435 # dirs == files without ".", short entries first 439 # dirs == files without ".", short entries first
436 $entries = [map $_->[0], 440 $entries = [map $_->[0],
441 my (@dirs, @nondirs); 445 my (@dirs, @nondirs);
442 446
443 my ($statcb, $schedcb); 447 my ($statcb, $schedcb);
444 my $nreq = 0; 448 my $nreq = 0;
445 449
450 my $statgrp = add $grp aio_group;
451
446 $schedcb = sub { 452 $schedcb = sub {
447 if (@$entries) { 453 if (@$entries) {
448 if ($nreq < $maxreq) { 454 if ($nreq < $maxreq) {
449 my $ent = pop @$entries; 455 my $ent = pop @$entries;
450 $nreq++; 456 $nreq++;
451 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) }; 457 add $statgrp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
452 } 458 }
453 } elsif (!$nreq) { 459 } elsif (!$nreq) {
454 # finished 460 # finished
461 $statgrp->cancel;
455 undef $statcb; 462 undef $statcb;
456 undef $schedcb; 463 undef $schedcb;
457 $cb->(\@dirs, \@nondirs) if $cb; 464 $grp->result (\@dirs, \@nondirs);
458 undef $cb;
459 } 465 }
460 }; 466 };
461 $statcb = sub { 467 $statcb = sub {
462 my ($status, $entry) = @_; 468 my ($status, $entry) = @_;
463 469
505callback with the fdatasync result code. 511callback with the fdatasync result code.
506 512
507If this call isn't available because your OS lacks it or it couldn't be 513If this call isn't available because your OS lacks it or it couldn't be
508detected, it will be emulated by calling C<fsync> instead. 514detected, it will be emulated by calling C<fsync> instead.
509 515
510=item aio_group $callback->() 516=item aio_group $callback->(...)
511 517
512[EXPERIMENTAL] 518[EXPERIMENTAL]
513 519
514This is a very special aio request: Instead of doing something, it is a 520This is a very special aio request: Instead of doing something, it is a
515container for other aio requests, which is useful if you want to bundle 521container for other aio requests, which is useful if you want to bundle
526 532
527 add $grp 533 add $grp
528 (aio_stat ...), 534 (aio_stat ...),
529 (aio_stat ...), 535 (aio_stat ...),
530 ...; 536 ...;
537
538=item aio_nop $callback->()
539
540This is a special request - it does nothing in itself and is only used for
541side effects, such as when you want to add a dummy request to a group so
542that finishing the requests in the group depends on executing the given
543code.
531 544
532=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED* 545=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
533 546
534Mainly used for debugging and benchmarking, this aio request puts one of 547Mainly used for debugging and benchmarking, this aio request puts one of
535the request workers to sleep for the given time. 548the request workers to sleep for the given time.
585You add requests by calling the C<add> method with one or more 598You add requests by calling the C<add> method with one or more
586C<IO::AIO::REQ> objects: 599C<IO::AIO::REQ> objects:
587 600
588 $grp->add (aio_unlink "..."); 601 $grp->add (aio_unlink "...");
589 602
590 add $grp aio_stat "...", sub { ... }; 603 add $grp aio_stat "...", sub {
604 $_[0] or return $grp->result ("error");
605
606 # add another request dynamically, if first succeeded
607 add $grp aio_open "...", sub {
608 $grp->result ("ok");
609 };
610 };
591 611
592This makes it very easy to create composite requests (see the source of 612This makes it very easy to create composite requests (see the source of
593C<aio_move> for an application) that work and feel like simple requests. 613C<aio_move> for an application) that work and feel like simple requests.
594 614
615=over 4
616
595The IO::AIO::GRP objects will be cleaned up during calls to 617=item * The IO::AIO::GRP objects will be cleaned up during calls to
596C<IO::AIO::poll_cb>, just like any other request. 618C<IO::AIO::poll_cb>, just like any other request.
597 619
598They can be canceled like any other request. Canceling will cancel not 620=item * They can be canceled like any other request. Canceling will cancel not
599just the request itself, but also all requests it contains. 621only the request itself, but also all requests it contains.
600 622
601They can also can also be added to other IO::AIO::GRP objects. 623=item * They can also can also be added to other IO::AIO::GRP objects.
624
625=item * You must not add requests to a group from within the group callback (or
626any later time).
627
628=item * This does not harmonise well with C<max_outstanding>, so best do
629not combine C<aio_group> with it. Groups and feeders are recommended for
630this kind of concurrency-limiting.
631
632=back
602 633
603Their lifetime, simplified, looks like this: when they are empty, they 634Their lifetime, simplified, looks like this: when they are empty, they
604will finish very quickly. If they contain only requests that are in the 635will finish very quickly. If they contain only requests that are in the
605C<done> state, they will also finish. Otherwise they will continue to 636C<done> state, they will also finish. Otherwise they will continue to
606exist. 637exist.
620be added, including other groups, as long as you do not create circular 651be added, including other groups, as long as you do not create circular
621dependencies. 652dependencies.
622 653
623Returns all its arguments. 654Returns all its arguments.
624 655
656=item $grp->result (...)
657
658Set the result value(s) that will be passed to the group callback when all
659subrequests have finished. By default, no argument will be passed.
660
661=item $grp->set_feeder ($callback->($grp))
662
663[VERY EXPERIMENTAL]
664
665Sets a feeder/generator on this group: every group can have an attached
666generator that generates requests if idle. The idea behind this is that,
667although you could just queue as many requests as you want in a group,
668this might starve other requests for a potentially long time. For
669example, C<aio_scandir> might generate hundreds of thousands C<aio_stat>
670requests, delaying any later requests for a long time.
671
672To avoid this, and allow incremental generation of requests, you can
673instead a group and set a feeder on it that generates those requests. The
674feeder will be called whenever there are few enough (see C<feeder_limit>,
675below) requests active in the group itself and is expected to queue more
676requests.
677
678The feeder can queue as many requests as it likes (i.e. C<add> does not
679impose any limits).
680
681If the feeder does not queue more requests when called, it will be
682automatically removed from the group.
683
684If the feeder limit is C<0>, it will be set to C<2> automatically.
685
686Example:
687
688 # stat all files in @files, but only ever use four aio requests concurrently:
689
690 my $grp = aio_group sub { print "finished\n" };
691 $grp->feeder_limit (4);
692 $grp->set_feeder (sub {
693 my $file = pop @files
694 or return;
695
696 add $grp aio_stat $file, sub { ... };
697 });
698
699=item $grp->feeder_limit ($num)
700
701Sets the feeder limit for the group: The feeder will be called whenever
702the group contains less than this many requests.
703
704Setting the limit to C<0> will pause the feeding process.
705
625=back 706=back
626 707
627=head2 SUPPORT FUNCTIONS 708=head2 SUPPORT FUNCTIONS
628 709
629=over 4 710=over 4
686 IO::AIO::poll_wait, IO::AIO::poll_cb 767 IO::AIO::poll_wait, IO::AIO::poll_cb
687 if IO::AIO::nreqs; 768 if IO::AIO::nreqs;
688 769
689=item IO::AIO::min_parallel $nthreads 770=item IO::AIO::min_parallel $nthreads
690 771
691Set the minimum number of AIO threads to C<$nthreads>. The current default 772Set the minimum number of AIO threads to C<$nthreads>. The current
692is C<4>, which means four asynchronous operations can be done at one time 773default is C<8>, which means eight asynchronous operations can execute
693(the number of outstanding operations, however, is unlimited). 774concurrently at any one time (the number of outstanding requests,
775however, is unlimited).
694 776
695IO::AIO starts threads only on demand, when an AIO request is queued and 777IO::AIO starts threads only on demand, when an AIO request is queued and
696no free thread exists. 778no free thread exists.
697 779
698It is recommended to keep the number of threads low, as some Linux 780It is recommended to keep the number of threads relatively low, as some
699kernel versions will scale negatively with the number of threads (higher 781Linux kernel versions will scale negatively with the number of threads
700parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32 782(higher parallelity => MUCH higher latency). With current Linux 2.6
701threads should be fine. 783versions, 4-32 threads should be fine.
702 784
703Under most circumstances you don't need to call this function, as the 785Under most circumstances you don't need to call this function, as the
704module selects a default that is suitable for low to moderate load. 786module selects a default that is suitable for low to moderate load.
705 787
706=item IO::AIO::max_parallel $nthreads 788=item IO::AIO::max_parallel $nthreads
717 799
718Under normal circumstances you don't need to call this function. 800Under normal circumstances you don't need to call this function.
719 801
720=item $oldnreqs = IO::AIO::max_outstanding $nreqs 802=item $oldnreqs = IO::AIO::max_outstanding $nreqs
721 803
804[DEPRECATED]
805
722Sets the maximum number of outstanding requests to C<$nreqs>. If you 806Sets the maximum number of outstanding requests to C<$nreqs>. If you
723try to queue up more than this number of requests, the caller will block until 807try to queue up more than this number of requests, the caller will block until
724some requests have been handled. 808some requests have been handled.
725 809
726The default is very large, so normally there is no practical limit. If you 810The default is very large, so normally there is no practical limit. If you
727queue up many requests in a loop it often improves speed if you set 811queue up many requests in a loop it often improves speed if you set
728this to a relatively low number, such as C<100>. 812this to a relatively low number, such as C<100>.
813
814This function does not work well together with C<aio_group>'s, and their
815feeder interface is better suited to limiting concurrency, so do not use
816this function.
729 817
730Under normal circumstances you don't need to call this function. 818Under normal circumstances you don't need to call this function.
731 819
732=back 820=back
733 821
747 or return undef; 835 or return undef;
748 836
749 *$sym 837 *$sym
750} 838}
751 839
752min_parallel 4; 840min_parallel 8;
753 841
754END { 842END {
755 max_parallel 0; 843 max_parallel 0;
756} 844}
757 845
771 859
772In short: the parent will, after a short pause, continue as if fork had 860In short: the parent will, after a short pause, continue as if fork had
773not been called, while the child will act as if IO::AIO has not been used 861not been called, while the child will act as if IO::AIO has not been used
774yet. 862yet.
775 863
864=head2 MEMORY USAGE
865
866Each aio request uses - depending on your architecture - around 128 bytes
867of memory. In addition, stat requests need a stat buffer (possibly a few
868hundred bytes). Perl scalars and other data passed into aio requests will
869also be locked.
870
871This is now awfully much, so queuing lots of requests is not usually a
872problem.
873
874Each thread needs a stack area which is usually around 16k, sometimes much
875larger, depending on the OS.
876
776=head1 SEE ALSO 877=head1 SEE ALSO
777 878
778L<Coro>, L<Linux::AIO> (obsolete). 879L<Coro>, L<Linux::AIO> (obsolete).
779 880
780=head1 AUTHOR 881=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines