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.67 by root, Tue Oct 24 02:25:16 2006 UTC

51 51
52This module implements asynchronous I/O using whatever means your 52This module implements asynchronous I/O using whatever means your
53operating system supports. 53operating system supports.
54 54
55Currently, a number of threads are started that execute your read/writes 55Currently, a number of threads are started that execute your read/writes
56and signal their completion. You don't need thread support in your libc or 56and signal their completion. You don't need thread support in perl, and
57perl, and the threads created by this module will not be visible to the 57the threads created by this module will not be visible to perl. In the
58pthreads library. In the future, this module might make use of the native 58future, this module might make use of the native aio functions available
59aio functions available on many operating systems. However, they are often 59on many operating systems. However, they are often not well-supported
60not well-supported (Linux doesn't allow them on normal files currently, 60(Linux doesn't allow them on normal files currently, for example),
61for example), and they would only support aio_read and aio_write, so the 61and they would only support aio_read and aio_write, so the remaining
62remaining functionality would have to be implemented using threads anyway. 62functionality would have to be implemented using threads anyway.
63 63
64Although the module will work with in the presence of other threads, it is 64Although the module will work with in the presence of other threads, it is
65currently not reentrant, so use appropriate locking yourself, always call 65currently not reentrant, so use appropriate locking yourself, always call
66C<poll_cb> from within the same thread, or never call C<poll_cb> (or other 66C<poll_cb> from within the same thread, or never call C<poll_cb> (or other
67C<aio_> functions) recursively. 67C<aio_> functions) recursively.
76use base 'Exporter'; 76use base 'Exporter';
77 77
78BEGIN { 78BEGIN {
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 @AIO_REQ = 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 = (@AIO_REQ, qw(aioreq_pri));
86 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush
87 min_parallel max_parallel max_outstanding nreqs);
86 88
87 @IO::AIO::GRP::ISA = 'IO::AIO::REQ'; 89 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
88 90
89 require XSLoader; 91 require XSLoader;
90 XSLoader::load ("IO::AIO", $VERSION); 92 XSLoader::load ("IO::AIO", $VERSION);
182 print "read $_[0] bytes: <$buffer>\n"; 184 print "read $_[0] bytes: <$buffer>\n";
183 }; 185 };
184 186
185=item aio_move $srcpath, $dstpath, $callback->($status) 187=item aio_move $srcpath, $dstpath, $callback->($status)
186 188
189[EXPERIMENTAL due to internal aio_group use]
190
187Try to move the I<file> (directories not supported as either source or 191Try to move the I<file> (directories not supported as either source or
188destination) from C<$srcpath> to C<$dstpath> and call the callback with 192destination) from C<$srcpath> to C<$dstpath> and call the callback with
189the C<0> (error) or C<-1> ok. 193the C<0> (error) or C<-1> ok.
190 194
191This is a composite request that tries to rename(2) the file first. If 195This is a composite request that tries to rename(2) the file first. If
201=cut 205=cut
202 206
203sub aio_move($$$) { 207sub aio_move($$$) {
204 my ($src, $dst, $cb) = @_; 208 my ($src, $dst, $cb) = @_;
205 209
206 my $grp = aio_group; 210 my $grp = aio_group $cb;
207 211
208 add $grp aio_rename $src, $dst, sub { 212 add $grp aio_rename $src, $dst, sub {
209 if ($_[0] && $! == EXDEV) { 213 if ($_[0] && $! == EXDEV) {
210 add $grp aio_open $src, O_RDONLY, 0, sub { 214 add $grp aio_open $src, O_RDONLY, 0, sub {
211 if (my $src_fh = $_[0]) { 215 if (my $src_fh = $_[0]) {
221 chmod $stat[2] & 07777, $dst_fh; 225 chmod $stat[2] & 07777, $dst_fh;
222 chown $stat[4], $stat[5], $dst_fh; 226 chown $stat[4], $stat[5], $dst_fh;
223 close $dst_fh; 227 close $dst_fh;
224 228
225 add $grp aio_unlink $src, sub { 229 add $grp aio_unlink $src, sub {
226 $cb->($_[0]); 230 $grp->result ($_[0]);
227 }; 231 };
228 } else { 232 } else {
229 my $errno = $!; 233 my $errno = $!;
230 add $grp aio_unlink $dst, sub { 234 add $grp aio_unlink $dst, sub {
231 $! = $errno; 235 $! = $errno;
232 $cb->(-1); 236 $grp->result (-1);
233 }; 237 };
234 } 238 }
235 }; 239 };
236 } else { 240 } else {
237 $cb->(-1); 241 $grp->result (-1);
238 } 242 }
239 }, 243 },
240 244
241 } else { 245 } else {
242 $cb->(-1); 246 $grp->result (-1);
243 } 247 }
244 }; 248 };
245 } else { 249 } else {
246 $cb->($_[0]); 250 $grp->result ($_[0]);
247 } 251 }
248 }; 252 };
249 253
250 $grp 254 $grp
251} 255}
343The callback a single argument which is either C<undef> or an array-ref 347The callback a single argument which is either C<undef> or an array-ref
344with the filenames. 348with the filenames.
345 349
346=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 350=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
347 351
352[EXPERIMENTAL due to internal aio_group use]
353
348Scans a directory (similar to C<aio_readdir>) but additionally tries to 354Scans a directory (similar to C<aio_readdir>) but additionally tries to
349separate the entries of directory C<$path> into two sets of names, ones 355separate 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 356you can recurse into (directories or links to them), and ones you cannot
351recurse into (everything else). 357recurse into (everything else).
352 358
353C<aio_scandir> is a composite request that consists of many sub 359C<aio_scandir> is a composite request that creates of many sub requests_
354requests. C<$maxreq> specifies the maximum number of outstanding aio 360C<$maxreq> specifies the maximum number of outstanding aio requests that
355requests that this function generates. If it is C<< <= 0 >>, then a 361this function generates. If it is C<< <= 0 >>, then a suitable default
356suitable default will be chosen (currently 8). 362will be chosen (currently 6).
357 363
358On error, the callback is called without arguments, otherwise it receives 364On error, the callback is called without arguments, otherwise it receives
359two array-refs with path-relative entry names. 365two array-refs with path-relative entry names.
360 366
361Example: 367Example:
398=cut 404=cut
399 405
400sub aio_scandir($$$) { 406sub aio_scandir($$$) {
401 my ($path, $maxreq, $cb) = @_; 407 my ($path, $maxreq, $cb) = @_;
402 408
403 my $grp = aio_group; 409 my $grp = aio_group $cb;
404 410
405 $maxreq = 8 if $maxreq <= 0; 411 $maxreq = 6 if $maxreq <= 0;
406 412
407 # stat once 413 # stat once
408 add $grp aio_stat $path, sub { 414 add $grp aio_stat $path, sub {
409 return $cb->() if $_[0]; 415 return $grp->result () if $_[0];
410 my $now = time; 416 my $now = time;
411 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 417 my $hash1 = join ":", (stat _)[0,1,3,7,9];
412 418
413 # read the directory entries 419 # read the directory entries
414 add $grp aio_readdir $path, sub { 420 add $grp aio_readdir $path, sub {
415 my $entries = shift 421 my $entries = shift
416 or return $cb->(); 422 or return $grp->result ();
417 423
418 # stat the dir another time 424 # stat the dir another time
419 add $grp aio_stat $path, sub { 425 add $grp aio_stat $path, sub {
420 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 426 my $hash2 = join ":", (stat _)[0,1,3,7,9];
421 427
426 $ndirs = -1; 432 $ndirs = -1;
427 } else { 433 } else {
428 # if nlink == 2, we are finished 434 # if nlink == 2, we are finished
429 # on non-posix-fs's, we rely on nlink < 2 435 # on non-posix-fs's, we rely on nlink < 2
430 $ndirs = (stat _)[3] - 2 436 $ndirs = (stat _)[3] - 2
431 or return $cb->([], $entries); 437 or return $grp->result ([], $entries);
432 } 438 }
433 439
434 # sort into likely dirs and likely nondirs 440 # sort into likely dirs and likely nondirs
435 # dirs == files without ".", short entries first 441 # dirs == files without ".", short entries first
436 $entries = [map $_->[0], 442 $entries = [map $_->[0],
441 my (@dirs, @nondirs); 447 my (@dirs, @nondirs);
442 448
443 my ($statcb, $schedcb); 449 my ($statcb, $schedcb);
444 my $nreq = 0; 450 my $nreq = 0;
445 451
452 my $statgrp = add $grp aio_group;
453
446 $schedcb = sub { 454 $schedcb = sub {
447 if (@$entries) { 455 if (@$entries) {
448 if ($nreq < $maxreq) { 456 if ($nreq < $maxreq) {
449 my $ent = pop @$entries; 457 my $ent = pop @$entries;
450 $nreq++; 458 $nreq++;
451 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) }; 459 add $statgrp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
452 } 460 }
453 } elsif (!$nreq) { 461 } elsif (!$nreq) {
454 # finished 462 # finished
463 $statgrp->cancel;
455 undef $statcb; 464 undef $statcb;
456 undef $schedcb; 465 undef $schedcb;
457 $cb->(\@dirs, \@nondirs) if $cb; 466 $grp->result (\@dirs, \@nondirs);
458 undef $cb;
459 } 467 }
460 }; 468 };
461 $statcb = sub { 469 $statcb = sub {
462 my ($status, $entry) = @_; 470 my ($status, $entry) = @_;
463 471
505callback with the fdatasync result code. 513callback with the fdatasync result code.
506 514
507If this call isn't available because your OS lacks it or it couldn't be 515If 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. 516detected, it will be emulated by calling C<fsync> instead.
509 517
510=item aio_group $callback->() 518=item aio_group $callback->(...)
511 519
512[EXPERIMENTAL] 520[EXPERIMENTAL]
513 521
514This is a very special aio request: Instead of doing something, it is a 522This 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 523container for other aio requests, which is useful if you want to bundle
526 534
527 add $grp 535 add $grp
528 (aio_stat ...), 536 (aio_stat ...),
529 (aio_stat ...), 537 (aio_stat ...),
530 ...; 538 ...;
539
540=item aio_nop $callback->()
541
542This is a special request - it does nothing in itself and is only used for
543side effects, such as when you want to add a dummy request to a group so
544that finishing the requests in the group depends on executing the given
545code.
546
547While this request does nothing, it still goes through the execution
548phase and still requires a worker thread. Thus, the callback will not
549be executed immediately but only after other requests in the queue have
550entered their execution phase. This can be used to measure request
551latency.
531 552
532=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED* 553=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
533 554
534Mainly used for debugging and benchmarking, this aio request puts one of 555Mainly used for debugging and benchmarking, this aio request puts one of
535the request workers to sleep for the given time. 556the request workers to sleep for the given time.
554callback) and B<done> (request has reached the end of its lifetime and 575callback) and B<done> (request has reached the end of its lifetime and
555holds no resources anymore). 576holds no resources anymore).
556 577
557=over 4 578=over 4
558 579
559=item $req->cancel 580=item cancel $req
560 581
561Cancels the request, if possible. Has the effect of skipping execution 582Cancels the request, if possible. Has the effect of skipping execution
562when entering the B<execute> state and skipping calling the callback when 583when entering the B<execute> state and skipping calling the callback when
563entering the the B<result> state, but will leave the request otherwise 584entering the the B<result> state, but will leave the request otherwise
564untouched. That means that requests that currently execute will not be 585untouched. That means that requests that currently execute will not be
565stopped and resources held by the request will not be freed prematurely. 586stopped and resources held by the request will not be freed prematurely.
566 587
588=item cb $req $callback->(...)
589
590Replace (or simply set) the callback registered to the request.
591
567=back 592=back
568 593
569=head2 IO::AIO::GRP CLASS 594=head2 IO::AIO::GRP CLASS
570 595
571This class is a subclass of L<IO::AIO::REQ>, so all its methods apply to 596This 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 610You add requests by calling the C<add> method with one or more
586C<IO::AIO::REQ> objects: 611C<IO::AIO::REQ> objects:
587 612
588 $grp->add (aio_unlink "..."); 613 $grp->add (aio_unlink "...");
589 614
590 add $grp aio_stat "...", sub { ... }; 615 add $grp aio_stat "...", sub {
616 $_[0] or return $grp->result ("error");
617
618 # add another request dynamically, if first succeeded
619 add $grp aio_open "...", sub {
620 $grp->result ("ok");
621 };
622 };
591 623
592This makes it very easy to create composite requests (see the source of 624This 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. 625C<aio_move> for an application) that work and feel like simple requests.
594 626
627=over 4
628
595The IO::AIO::GRP objects will be cleaned up during calls to 629=item * The IO::AIO::GRP objects will be cleaned up during calls to
596C<IO::AIO::poll_cb>, just like any other request. 630C<IO::AIO::poll_cb>, just like any other request.
597 631
598They can be canceled like any other request. Canceling will cancel not 632=item * They can be canceled like any other request. Canceling will cancel not
599just the request itself, but also all requests it contains. 633only the request itself, but also all requests it contains.
600 634
601They can also can also be added to other IO::AIO::GRP objects. 635=item * They can also can also be added to other IO::AIO::GRP objects.
636
637=item * You must not add requests to a group from within the group callback (or
638any later time).
639
640=item * This does not harmonise well with C<max_outstanding>, so best do
641not combine C<aio_group> with it. Groups and feeders are recommended for
642this kind of concurrency-limiting.
643
644=back
602 645
603Their lifetime, simplified, looks like this: when they are empty, they 646Their lifetime, simplified, looks like this: when they are empty, they
604will finish very quickly. If they contain only requests that are in the 647will finish very quickly. If they contain only requests that are in the
605C<done> state, they will also finish. Otherwise they will continue to 648C<done> state, they will also finish. Otherwise they will continue to
606exist. 649exist.
610group. And only when all those requests have finished will the the group 653group. And only when all those requests have finished will the the group
611itself finish. 654itself finish.
612 655
613=over 4 656=over 4
614 657
658=item add $grp ...
659
615=item $grp->add (...) 660=item $grp->add (...)
616
617=item add $grp ...
618 661
619Add one or more requests to the group. Any type of L<IO::AIO::REQ> can 662Add 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 663be added, including other groups, as long as you do not create circular
621dependencies. 664dependencies.
622 665
623Returns all its arguments. 666Returns all its arguments.
667
668=item $grp->result (...)
669
670Set the result value(s) that will be passed to the group callback when all
671subrequests have finished. By default, no argument will be passed.
672
673=item feed $grp $callback->($grp)
674
675[VERY EXPERIMENTAL]
676
677Sets a feeder/generator on this group: every group can have an attached
678generator that generates requests if idle. The idea behind this is that,
679although you could just queue as many requests as you want in a group,
680this might starve other requests for a potentially long time. For
681example, C<aio_scandir> might generate hundreds of thousands C<aio_stat>
682requests, delaying any later requests for a long time.
683
684To avoid this, and allow incremental generation of requests, you can
685instead a group and set a feeder on it that generates those requests. The
686feed callback will be called whenever there are few enough (see C<feed_limit>,
687below) requests active in the group itself and is expected to queue more
688requests.
689
690The feed can queue as many requests as it likes (i.e. C<add> does not
691impose any limits).
692
693If the feed does not queue more requests when called, it will be
694automatically removed from the group.
695
696If the feed limit is C<0>, it will be set to C<2> automatically.
697
698Example:
699
700 # stat all files in @files, but only ever use four aio requests concurrently:
701
702 my $grp = aio_group sub { print "finished\n" };
703 feed_limit $grp 4;
704 feed $grp sub {
705 my $file = pop @files
706 or return;
707
708 add $grp aio_stat $file, sub { ... };
709 };
710
711=item feed_limit $grp $num
712
713Sets the feeder limit for the group: The feeder will be called whenever
714the group contains less than this many requests.
715
716Setting the limit to C<0> will pause the feeding process.
624 717
625=back 718=back
626 719
627=head2 SUPPORT FUNCTIONS 720=head2 SUPPORT FUNCTIONS
628 721
686 IO::AIO::poll_wait, IO::AIO::poll_cb 779 IO::AIO::poll_wait, IO::AIO::poll_cb
687 if IO::AIO::nreqs; 780 if IO::AIO::nreqs;
688 781
689=item IO::AIO::min_parallel $nthreads 782=item IO::AIO::min_parallel $nthreads
690 783
691Set the minimum number of AIO threads to C<$nthreads>. The current default 784Set 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 785default is C<8>, which means eight asynchronous operations can execute
693(the number of outstanding operations, however, is unlimited). 786concurrently at any one time (the number of outstanding requests,
787however, is unlimited).
694 788
695IO::AIO starts threads only on demand, when an AIO request is queued and 789IO::AIO starts threads only on demand, when an AIO request is queued and
696no free thread exists. 790no free thread exists.
697 791
698It is recommended to keep the number of threads low, as some Linux 792It is recommended to keep the number of threads relatively low, as some
699kernel versions will scale negatively with the number of threads (higher 793Linux kernel versions will scale negatively with the number of threads
700parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32 794(higher parallelity => MUCH higher latency). With current Linux 2.6
701threads should be fine. 795versions, 4-32 threads should be fine.
702 796
703Under most circumstances you don't need to call this function, as the 797Under most circumstances you don't need to call this function, as the
704module selects a default that is suitable for low to moderate load. 798module selects a default that is suitable for low to moderate load.
705 799
706=item IO::AIO::max_parallel $nthreads 800=item IO::AIO::max_parallel $nthreads
717 811
718Under normal circumstances you don't need to call this function. 812Under normal circumstances you don't need to call this function.
719 813
720=item $oldnreqs = IO::AIO::max_outstanding $nreqs 814=item $oldnreqs = IO::AIO::max_outstanding $nreqs
721 815
816[DEPRECATED]
817
722Sets the maximum number of outstanding requests to C<$nreqs>. If you 818Sets 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 819try to queue up more than this number of requests, the caller will block until
724some requests have been handled. 820some requests have been handled.
725 821
726The default is very large, so normally there is no practical limit. If you 822The 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 823queue up many requests in a loop it often improves speed if you set
728this to a relatively low number, such as C<100>. 824this to a relatively low number, such as C<100>.
825
826This function does not work well together with C<aio_group>'s, and their
827feeder interface is better suited to limiting concurrency, so do not use
828this function.
729 829
730Under normal circumstances you don't need to call this function. 830Under normal circumstances you don't need to call this function.
731 831
732=back 832=back
733 833
747 or return undef; 847 or return undef;
748 848
749 *$sym 849 *$sym
750} 850}
751 851
752min_parallel 4; 852min_parallel 8;
753 853
754END { 854END {
755 max_parallel 0; 855 max_parallel 0;
756} 856}
757 857
771 871
772In short: the parent will, after a short pause, continue as if fork had 872In 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 873not been called, while the child will act as if IO::AIO has not been used
774yet. 874yet.
775 875
876=head2 MEMORY USAGE
877
878Each aio request uses - depending on your architecture - around 128 bytes
879of memory. In addition, stat requests need a stat buffer (possibly a few
880hundred bytes). Perl scalars and other data passed into aio requests will
881also be locked.
882
883This is now awfully much, so queuing lots of requests is not usually a
884problem.
885
886Each thread needs a stack area which is usually around 16k, sometimes much
887larger, depending on the OS.
888
776=head1 SEE ALSO 889=head1 SEE ALSO
777 890
778L<Coro>, L<Linux::AIO> (obsolete). 891L<Coro>, L<Linux::AIO> (obsolete).
779 892
780=head1 AUTHOR 893=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines