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.65 by root, Mon Oct 23 14:49:51 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.
544
545While this request does nothing, it still goes through the execution
546phase and still requires a worker thread. Thus, the callback will not
547be executed immediately but only after other requests in the queue have
548entered their execution phase. This can be used to measure request
549latency.
531 550
532=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED* 551=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
533 552
534Mainly used for debugging and benchmarking, this aio request puts one of 553Mainly used for debugging and benchmarking, this aio request puts one of
535the request workers to sleep for the given time. 554the request workers to sleep for the given time.
554callback) and B<done> (request has reached the end of its lifetime and 573callback) and B<done> (request has reached the end of its lifetime and
555holds no resources anymore). 574holds no resources anymore).
556 575
557=over 4 576=over 4
558 577
559=item $req->cancel 578=item cancel $req
560 579
561Cancels the request, if possible. Has the effect of skipping execution 580Cancels the request, if possible. Has the effect of skipping execution
562when entering the B<execute> state and skipping calling the callback when 581when entering the B<execute> state and skipping calling the callback when
563entering the the B<result> state, but will leave the request otherwise 582entering the the B<result> state, but will leave the request otherwise
564untouched. That means that requests that currently execute will not be 583untouched. That means that requests that currently execute will not be
565stopped and resources held by the request will not be freed prematurely. 584stopped and resources held by the request will not be freed prematurely.
566 585
586=item cb $req $callback->(...)
587
588Replace (or simply set) the callback registered to the request.
589
567=back 590=back
568 591
569=head2 IO::AIO::GRP CLASS 592=head2 IO::AIO::GRP CLASS
570 593
571This class is a subclass of L<IO::AIO::REQ>, so all its methods apply to 594This class is a subclass of L<IO::AIO::REQ>, so all its methods apply to
585You add requests by calling the C<add> method with one or more 608You add requests by calling the C<add> method with one or more
586C<IO::AIO::REQ> objects: 609C<IO::AIO::REQ> objects:
587 610
588 $grp->add (aio_unlink "..."); 611 $grp->add (aio_unlink "...");
589 612
590 add $grp aio_stat "...", sub { ... }; 613 add $grp aio_stat "...", sub {
614 $_[0] or return $grp->result ("error");
615
616 # add another request dynamically, if first succeeded
617 add $grp aio_open "...", sub {
618 $grp->result ("ok");
619 };
620 };
591 621
592This makes it very easy to create composite requests (see the source of 622This 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. 623C<aio_move> for an application) that work and feel like simple requests.
594 624
625=over 4
626
595The IO::AIO::GRP objects will be cleaned up during calls to 627=item * The IO::AIO::GRP objects will be cleaned up during calls to
596C<IO::AIO::poll_cb>, just like any other request. 628C<IO::AIO::poll_cb>, just like any other request.
597 629
598They can be canceled like any other request. Canceling will cancel not 630=item * They can be canceled like any other request. Canceling will cancel not
599just the request itself, but also all requests it contains. 631only the request itself, but also all requests it contains.
600 632
601They can also can also be added to other IO::AIO::GRP objects. 633=item * They can also can also be added to other IO::AIO::GRP objects.
634
635=item * You must not add requests to a group from within the group callback (or
636any later time).
637
638=item * This does not harmonise well with C<max_outstanding>, so best do
639not combine C<aio_group> with it. Groups and feeders are recommended for
640this kind of concurrency-limiting.
641
642=back
602 643
603Their lifetime, simplified, looks like this: when they are empty, they 644Their lifetime, simplified, looks like this: when they are empty, they
604will finish very quickly. If they contain only requests that are in the 645will finish very quickly. If they contain only requests that are in the
605C<done> state, they will also finish. Otherwise they will continue to 646C<done> state, they will also finish. Otherwise they will continue to
606exist. 647exist.
610group. And only when all those requests have finished will the the group 651group. And only when all those requests have finished will the the group
611itself finish. 652itself finish.
612 653
613=over 4 654=over 4
614 655
656=item add $grp ...
657
615=item $grp->add (...) 658=item $grp->add (...)
616
617=item add $grp ...
618 659
619Add one or more requests to the group. Any type of L<IO::AIO::REQ> can 660Add one or more requests to the group. Any type of L<IO::AIO::REQ> can
620be added, including other groups, as long as you do not create circular 661be added, including other groups, as long as you do not create circular
621dependencies. 662dependencies.
622 663
623Returns all its arguments. 664Returns all its arguments.
665
666=item $grp->result (...)
667
668Set the result value(s) that will be passed to the group callback when all
669subrequests have finished. By default, no argument will be passed.
670
671=item feed $grp $callback->($grp)
672
673[VERY EXPERIMENTAL]
674
675Sets a feeder/generator on this group: every group can have an attached
676generator that generates requests if idle. The idea behind this is that,
677although you could just queue as many requests as you want in a group,
678this might starve other requests for a potentially long time. For
679example, C<aio_scandir> might generate hundreds of thousands C<aio_stat>
680requests, delaying any later requests for a long time.
681
682To avoid this, and allow incremental generation of requests, you can
683instead a group and set a feeder on it that generates those requests. The
684feed callback will be called whenever there are few enough (see C<feed_limit>,
685below) requests active in the group itself and is expected to queue more
686requests.
687
688The feed can queue as many requests as it likes (i.e. C<add> does not
689impose any limits).
690
691If the feed does not queue more requests when called, it will be
692automatically removed from the group.
693
694If the feed limit is C<0>, it will be set to C<2> automatically.
695
696Example:
697
698 # stat all files in @files, but only ever use four aio requests concurrently:
699
700 my $grp = aio_group sub { print "finished\n" };
701 feed_limit $grp 4;
702 feed $grp sub {
703 my $file = pop @files
704 or return;
705
706 add $grp aio_stat $file, sub { ... };
707 };
708
709=item feed_limit $grp $num
710
711Sets the feeder limit for the group: The feeder will be called whenever
712the group contains less than this many requests.
713
714Setting the limit to C<0> will pause the feeding process.
624 715
625=back 716=back
626 717
627=head2 SUPPORT FUNCTIONS 718=head2 SUPPORT FUNCTIONS
628 719
686 IO::AIO::poll_wait, IO::AIO::poll_cb 777 IO::AIO::poll_wait, IO::AIO::poll_cb
687 if IO::AIO::nreqs; 778 if IO::AIO::nreqs;
688 779
689=item IO::AIO::min_parallel $nthreads 780=item IO::AIO::min_parallel $nthreads
690 781
691Set the minimum number of AIO threads to C<$nthreads>. The current default 782Set 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 783default is C<8>, which means eight asynchronous operations can execute
693(the number of outstanding operations, however, is unlimited). 784concurrently at any one time (the number of outstanding requests,
785however, is unlimited).
694 786
695IO::AIO starts threads only on demand, when an AIO request is queued and 787IO::AIO starts threads only on demand, when an AIO request is queued and
696no free thread exists. 788no free thread exists.
697 789
698It is recommended to keep the number of threads low, as some Linux 790It is recommended to keep the number of threads relatively low, as some
699kernel versions will scale negatively with the number of threads (higher 791Linux kernel versions will scale negatively with the number of threads
700parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32 792(higher parallelity => MUCH higher latency). With current Linux 2.6
701threads should be fine. 793versions, 4-32 threads should be fine.
702 794
703Under most circumstances you don't need to call this function, as the 795Under most circumstances you don't need to call this function, as the
704module selects a default that is suitable for low to moderate load. 796module selects a default that is suitable for low to moderate load.
705 797
706=item IO::AIO::max_parallel $nthreads 798=item IO::AIO::max_parallel $nthreads
717 809
718Under normal circumstances you don't need to call this function. 810Under normal circumstances you don't need to call this function.
719 811
720=item $oldnreqs = IO::AIO::max_outstanding $nreqs 812=item $oldnreqs = IO::AIO::max_outstanding $nreqs
721 813
814[DEPRECATED]
815
722Sets the maximum number of outstanding requests to C<$nreqs>. If you 816Sets 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 817try to queue up more than this number of requests, the caller will block until
724some requests have been handled. 818some requests have been handled.
725 819
726The default is very large, so normally there is no practical limit. If you 820The 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 821queue up many requests in a loop it often improves speed if you set
728this to a relatively low number, such as C<100>. 822this to a relatively low number, such as C<100>.
823
824This function does not work well together with C<aio_group>'s, and their
825feeder interface is better suited to limiting concurrency, so do not use
826this function.
729 827
730Under normal circumstances you don't need to call this function. 828Under normal circumstances you don't need to call this function.
731 829
732=back 830=back
733 831
747 or return undef; 845 or return undef;
748 846
749 *$sym 847 *$sym
750} 848}
751 849
752min_parallel 4; 850min_parallel 8;
753 851
754END { 852END {
755 max_parallel 0; 853 max_parallel 0;
756} 854}
757 855
771 869
772In short: the parent will, after a short pause, continue as if fork had 870In 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 871not been called, while the child will act as if IO::AIO has not been used
774yet. 872yet.
775 873
874=head2 MEMORY USAGE
875
876Each aio request uses - depending on your architecture - around 128 bytes
877of memory. In addition, stat requests need a stat buffer (possibly a few
878hundred bytes). Perl scalars and other data passed into aio requests will
879also be locked.
880
881This is now awfully much, so queuing lots of requests is not usually a
882problem.
883
884Each thread needs a stack area which is usually around 16k, sometimes much
885larger, depending on the OS.
886
776=head1 SEE ALSO 887=head1 SEE ALSO
777 888
778L<Coro>, L<Linux::AIO> (obsolete). 889L<Coro>, L<Linux::AIO> (obsolete).
779 890
780=head1 AUTHOR 891=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines