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.56 by root, Sun Oct 22 00:53:47 2006 UTC vs.
Revision 1.71 by root, Tue Oct 24 11:57:30 2006 UTC

18 }; 18 };
19 19
20 # version 2+ has request and group objects 20 # version 2+ has request and group objects
21 use IO::AIO 2; 21 use IO::AIO 2;
22 22
23 aioreq_pri 4; # give next request a very high priority
23 my $req = aio_unlink "/tmp/file", sub { }; 24 my $req = aio_unlink "/tmp/file", sub { };
24 $req->cancel; # cancel request if still in queue 25 $req->cancel; # cancel request if still in queue
25 26
26 my $grp = aio_group sub { print "all stats done\n" }; 27 my $grp = aio_group sub { print "all stats done\n" };
27 add $grp aio_stat "..." for ...; 28 add $grp aio_stat "..." for ...;
51 52
52This module implements asynchronous I/O using whatever means your 53This module implements asynchronous I/O using whatever means your
53operating system supports. 54operating system supports.
54 55
55Currently, a number of threads are started that execute your read/writes 56Currently, 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 57and 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 58the 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 59future, this module might make use of the native aio functions available
59aio functions available on many operating systems. However, they are often 60on many operating systems. However, they are often not well-supported
60not well-supported (Linux doesn't allow them on normal files currently, 61(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 62and they would only support aio_read and aio_write, so the remaining
62remaining functionality would have to be implemented using threads anyway. 63functionality would have to be implemented using threads anyway.
63 64
64Although the module will work with in the presence of other threads, it is 65Although the module will work with in the presence of other threads,
65currently not reentrant, so use appropriate locking yourself, always call 66it is currently not reentrant in any way, so use appropriate locking
66C<poll_cb> from within the same thread, or never call C<poll_cb> (or other 67yourself, always call C<poll_cb> from within the same thread, or never
67C<aio_> functions) recursively. 68call C<poll_cb> (or other C<aio_> functions) recursively.
68 69
69=cut 70=cut
70 71
71package IO::AIO; 72package IO::AIO;
72 73
76use base 'Exporter'; 77use base 'Exporter';
77 78
78BEGIN { 79BEGIN {
79 our $VERSION = '2.0'; 80 our $VERSION = '2.0';
80 81
81 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 82 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 83 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 84 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
84 aio_group); 85 aio_group aio_nop);
85 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs); 86 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice));
87 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush
88 min_parallel max_parallel max_outstanding nreqs);
86 89
87 @IO::AIO::GRP::ISA = 'IO::AIO::REQ'; 90 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
88 91
89 require XSLoader; 92 require XSLoader;
90 XSLoader::load ("IO::AIO", $VERSION); 93 XSLoader::load ("IO::AIO", $VERSION);
121environment, d) use Glib::filename_from_unicode on unicode filenames or e) 124environment, d) use Glib::filename_from_unicode on unicode filenames or e)
122use something else. 125use something else.
123 126
124=over 4 127=over 4
125 128
129=item aioreq_pri $pri
130
131Sets the priority for the next aio request. The default priority
132is C<0>, the minimum and maximum priorities are C<-4> and C<4>,
133respectively. Requests with higher priority will be serviced first.
134
135The priority will be reset to C<0> after each call to one of the C<aio_>
136functions.
137
138Example: open a file with low priority, then read something from it with
139higher priority so the read request is serviced before other low priority
140open requests (potentially spamming the cache):
141
142 aioreq_pri -3;
143 aio_open ..., sub {
144 return unless $_[0];
145
146 aioreq_pri -2;
147 aio_read $_[0], ..., sub {
148 ...
149 };
150 };
151
152=item aioreq_nice $pri_adjust
153
154Similar to C<aioreq_pri>, but subtracts the given value from the current
155priority, so effects are cumulative.
156
126=item aio_open $pathname, $flags, $mode, $callback->($fh) 157=item aio_open $pathname, $flags, $mode, $callback->($fh)
127 158
128Asynchronously open or create a file and call the callback with a newly 159Asynchronously open or create a file and call the callback with a newly
129created filehandle for the file. 160created filehandle for the file.
130 161
201=cut 232=cut
202 233
203sub aio_move($$$) { 234sub aio_move($$$) {
204 my ($src, $dst, $cb) = @_; 235 my ($src, $dst, $cb) = @_;
205 236
206 my $grp = aio_group; 237 my $grp = aio_group $cb;
207 238
208 add $grp aio_rename $src, $dst, sub { 239 add $grp aio_rename $src, $dst, sub {
209 if ($_[0] && $! == EXDEV) { 240 if ($_[0] && $! == EXDEV) {
210 add $grp aio_open $src, O_RDONLY, 0, sub { 241 add $grp aio_open $src, O_RDONLY, 0, sub {
211 if (my $src_fh = $_[0]) { 242 if (my $src_fh = $_[0]) {
221 chmod $stat[2] & 07777, $dst_fh; 252 chmod $stat[2] & 07777, $dst_fh;
222 chown $stat[4], $stat[5], $dst_fh; 253 chown $stat[4], $stat[5], $dst_fh;
223 close $dst_fh; 254 close $dst_fh;
224 255
225 add $grp aio_unlink $src, sub { 256 add $grp aio_unlink $src, sub {
226 $cb->($_[0]); 257 $grp->result ($_[0]);
227 }; 258 };
228 } else { 259 } else {
229 my $errno = $!; 260 my $errno = $!;
230 add $grp aio_unlink $dst, sub { 261 add $grp aio_unlink $dst, sub {
231 $! = $errno; 262 $! = $errno;
232 $cb->(-1); 263 $grp->result (-1);
233 }; 264 };
234 } 265 }
235 }; 266 };
236 } else { 267 } else {
237 $cb->(-1); 268 $grp->result (-1);
238 } 269 }
239 }, 270 },
240 271
241 } else { 272 } else {
242 $cb->(-1); 273 $grp->result (-1);
243 } 274 }
244 }; 275 };
245 } else { 276 } else {
246 $cb->($_[0]); 277 $grp->result ($_[0]);
247 } 278 }
248 }; 279 };
249 280
250 $grp 281 $grp
251} 282}
348Scans a directory (similar to C<aio_readdir>) but additionally tries to 379Scans a directory (similar to C<aio_readdir>) but additionally tries to
349separate the entries of directory C<$path> into two sets of names, ones 380separate 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 381you can recurse into (directories or links to them), and ones you cannot
351recurse into (everything else). 382recurse into (everything else).
352 383
353C<aio_scandir> is a composite request that consists of many sub 384C<aio_scandir> is a composite request that creates of many sub requests_
354requests. C<$maxreq> specifies the maximum number of outstanding aio 385C<$maxreq> specifies the maximum number of outstanding aio requests that
355requests that this function generates. If it is C<< <= 0 >>, then a 386this function generates. If it is C<< <= 0 >>, then a suitable default
356suitable default will be chosen (currently 8). 387will be chosen (currently 6).
357 388
358On error, the callback is called without arguments, otherwise it receives 389On error, the callback is called without arguments, otherwise it receives
359two array-refs with path-relative entry names. 390two array-refs with path-relative entry names.
360 391
361Example: 392Example:
398=cut 429=cut
399 430
400sub aio_scandir($$$) { 431sub aio_scandir($$$) {
401 my ($path, $maxreq, $cb) = @_; 432 my ($path, $maxreq, $cb) = @_;
402 433
403 my $grp = aio_group; 434 my $grp = aio_group $cb;
404 435
405 $maxreq = 8 if $maxreq <= 0; 436 $maxreq = 6 if $maxreq <= 0;
406 437
407 # stat once 438 # stat once
408 add $grp aio_stat $path, sub { 439 add $grp aio_stat $path, sub {
409 return $cb->() if $_[0]; 440 return $grp->result () if $_[0];
410 my $now = time; 441 my $now = time;
411 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 442 my $hash1 = join ":", (stat _)[0,1,3,7,9];
412 443
413 # read the directory entries 444 # read the directory entries
414 add $grp aio_readdir $path, sub { 445 add $grp aio_readdir $path, sub {
415 my $entries = shift 446 my $entries = shift
416 or return $cb->(); 447 or return $grp->result ();
417 448
418 # stat the dir another time 449 # stat the dir another time
419 add $grp aio_stat $path, sub { 450 add $grp aio_stat $path, sub {
420 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 451 my $hash2 = join ":", (stat _)[0,1,3,7,9];
421 452
426 $ndirs = -1; 457 $ndirs = -1;
427 } else { 458 } else {
428 # if nlink == 2, we are finished 459 # if nlink == 2, we are finished
429 # on non-posix-fs's, we rely on nlink < 2 460 # on non-posix-fs's, we rely on nlink < 2
430 $ndirs = (stat _)[3] - 2 461 $ndirs = (stat _)[3] - 2
431 or return $cb->([], $entries); 462 or return $grp->result ([], $entries);
432 } 463 }
433 464
434 # sort into likely dirs and likely nondirs 465 # sort into likely dirs and likely nondirs
435 # dirs == files without ".", short entries first 466 # dirs == files without ".", short entries first
436 $entries = [map $_->[0], 467 $entries = [map $_->[0],
441 my (@dirs, @nondirs); 472 my (@dirs, @nondirs);
442 473
443 my ($statcb, $schedcb); 474 my ($statcb, $schedcb);
444 my $nreq = 0; 475 my $nreq = 0;
445 476
477 my $statgrp = add $grp aio_group;
478
446 $schedcb = sub { 479 $schedcb = sub {
447 if (@$entries) { 480 if (@$entries) {
448 if ($nreq < $maxreq) { 481 if ($nreq < $maxreq) {
449 my $ent = pop @$entries; 482 my $ent = pop @$entries;
450 $nreq++; 483 $nreq++;
451 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) }; 484 add $statgrp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
452 } 485 }
453 } elsif (!$nreq) { 486 } elsif (!$nreq) {
454 # finished 487 # finished
488 $statgrp->cancel;
455 undef $statcb; 489 undef $statcb;
456 undef $schedcb; 490 undef $schedcb;
457 $cb->(\@dirs, \@nondirs) if $cb; 491 $grp->result (\@dirs, \@nondirs);
458 undef $cb;
459 } 492 }
460 }; 493 };
461 $statcb = sub { 494 $statcb = sub {
462 my ($status, $entry) = @_; 495 my ($status, $entry) = @_;
463 496
505callback with the fdatasync result code. 538callback with the fdatasync result code.
506 539
507If this call isn't available because your OS lacks it or it couldn't be 540If 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. 541detected, it will be emulated by calling C<fsync> instead.
509 542
510=item aio_group $callback->() 543=item aio_group $callback->(...)
511 544
512[EXPERIMENTAL] 545[EXPERIMENTAL]
513 546
514This is a very special aio request: Instead of doing something, it is a 547This 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 548container for other aio requests, which is useful if you want to bundle
516many requests into a single, composite, request. 549many requests into a single, composite, request with a definite callback
550and the ability to cancel the whole request with its subrequests.
517 551
518Returns an object of class L<IO::AIO::GRP>. See its documentation below 552Returns an object of class L<IO::AIO::GRP>. See its documentation below
519for more info. 553for more info.
520 554
521Example: 555Example:
527 add $grp 561 add $grp
528 (aio_stat ...), 562 (aio_stat ...),
529 (aio_stat ...), 563 (aio_stat ...),
530 ...; 564 ...;
531 565
566=item aio_nop $callback->()
567
568This is a special request - it does nothing in itself and is only used for
569side effects, such as when you want to add a dummy request to a group so
570that finishing the requests in the group depends on executing the given
571code.
572
573While this request does nothing, it still goes through the execution
574phase and still requires a worker thread. Thus, the callback will not
575be executed immediately but only after other requests in the queue have
576entered their execution phase. This can be used to measure request
577latency.
578
532=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED* 579=item IO::AIO::aio_busy $fractional_seconds, $callback->() *NOT EXPORTED*
533 580
534Mainly used for debugging and benchmarking, this aio request puts one of 581Mainly used for debugging and benchmarking, this aio request puts one of
535the request workers to sleep for the given time. 582the request workers to sleep for the given time.
536 583
537While it is theoretically handy to have simple I/O scheduling requests 584While it is theoretically handy to have simple I/O scheduling requests
538like sleep and file handle readable/writable, the overhead this creates 585like sleep and file handle readable/writable, the overhead this creates is
539is immense, so do not use this function except to put your application 586immense (it blocks a thread for a long time) so do not use this function
540under artificial I/O pressure. 587except to put your application under artificial I/O pressure.
541 588
542=back 589=back
543 590
544=head2 IO::AIO::REQ CLASS 591=head2 IO::AIO::REQ CLASS
545 592
554callback) and B<done> (request has reached the end of its lifetime and 601callback) and B<done> (request has reached the end of its lifetime and
555holds no resources anymore). 602holds no resources anymore).
556 603
557=over 4 604=over 4
558 605
559=item $req->cancel 606=item cancel $req
560 607
561Cancels the request, if possible. Has the effect of skipping execution 608Cancels the request, if possible. Has the effect of skipping execution
562when entering the B<execute> state and skipping calling the callback when 609when entering the B<execute> state and skipping calling the callback when
563entering the the B<result> state, but will leave the request otherwise 610entering the the B<result> state, but will leave the request otherwise
564untouched. That means that requests that currently execute will not be 611untouched. That means that requests that currently execute will not be
565stopped and resources held by the request will not be freed prematurely. 612stopped and resources held by the request will not be freed prematurely.
566 613
614=item cb $req $callback->(...)
615
616Replace (or simply set) the callback registered to the request.
617
567=back 618=back
568 619
569=head2 IO::AIO::GRP CLASS 620=head2 IO::AIO::GRP CLASS
570 621
571This class is a subclass of L<IO::AIO::REQ>, so all its methods apply to 622This 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 636You add requests by calling the C<add> method with one or more
586C<IO::AIO::REQ> objects: 637C<IO::AIO::REQ> objects:
587 638
588 $grp->add (aio_unlink "..."); 639 $grp->add (aio_unlink "...");
589 640
590 add $grp aio_stat "...", sub { ... }; 641 add $grp aio_stat "...", sub {
642 $_[0] or return $grp->result ("error");
643
644 # add another request dynamically, if first succeeded
645 add $grp aio_open "...", sub {
646 $grp->result ("ok");
647 };
648 };
591 649
592This makes it very easy to create composite requests (see the source of 650This 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. 651C<aio_move> for an application) that work and feel like simple requests.
594 652
653=over 4
654
595The IO::AIO::GRP objects will be cleaned up during calls to 655=item * The IO::AIO::GRP objects will be cleaned up during calls to
596C<IO::AIO::poll_cb>, just like any other request. 656C<IO::AIO::poll_cb>, just like any other request.
597 657
598They can be canceled like any other request. Canceling will cancel not 658=item * They can be canceled like any other request. Canceling will cancel not
599just the request itself, but also all requests it contains. 659only the request itself, but also all requests it contains.
600 660
601They can also can also be added to other IO::AIO::GRP objects. 661=item * They can also can also be added to other IO::AIO::GRP objects.
662
663=item * You must not add requests to a group from within the group callback (or
664any later time).
665
666=item * This does not harmonise well with C<max_outstanding>, so best do
667not combine C<aio_group> with it. Groups and feeders are recommended for
668this kind of concurrency-limiting.
669
670=back
602 671
603Their lifetime, simplified, looks like this: when they are empty, they 672Their lifetime, simplified, looks like this: when they are empty, they
604will finish very quickly. If they contain only requests that are in the 673will finish very quickly. If they contain only requests that are in the
605C<done> state, they will also finish. Otherwise they will continue to 674C<done> state, they will also finish. Otherwise they will continue to
606exist. 675exist.
607 676
677That means after creating a group you have some time to add requests. And
678in the callbacks of those requests, you can add further requests to the
679group. And only when all those requests have finished will the the group
680itself finish.
681
608=over 4 682=over 4
609 683
684=item add $grp ...
685
610=item $grp->add (...) 686=item $grp->add (...)
611 687
612=item add $grp ... 688Add one or more requests to the group. Any type of L<IO::AIO::REQ> can
689be added, including other groups, as long as you do not create circular
690dependencies.
613 691
614Add one or more 692Returns all its arguments.
615Cancels the request, if possible. Has the effect of skipping execution 693
616when entering the B<execute> state and skipping calling the callback when 694=item $grp->result (...)
617entering the the B<result> state, but will leave the request otherwise 695
618untouched. That means that requests that currently execute will not be 696Set the result value(s) that will be passed to the group callback when all
619stopped and resources held by the request will not be freed prematurely. 697subrequests have finished. By default, no argument will be passed.
698
699=item feed $grp $callback->($grp)
700
701[VERY EXPERIMENTAL]
702
703Sets a feeder/generator on this group: every group can have an attached
704generator that generates requests if idle. The idea behind this is that,
705although you could just queue as many requests as you want in a group,
706this might starve other requests for a potentially long time. For
707example, C<aio_scandir> might generate hundreds of thousands C<aio_stat>
708requests, delaying any later requests for a long time.
709
710To avoid this, and allow incremental generation of requests, you can
711instead a group and set a feeder on it that generates those requests. The
712feed callback will be called whenever there are few enough (see C<limit>,
713below) requests active in the group itself and is expected to queue more
714requests.
715
716The feed callback can queue as many requests as it likes (i.e. C<add> does
717not impose any limits).
718
719If the feed does not queue more requests when called, it will be
720automatically removed from the group.
721
722If the feed limit is C<0>, it will be set to C<2> automatically.
723
724Example:
725
726 # stat all files in @files, but only ever use four aio requests concurrently:
727
728 my $grp = aio_group sub { print "finished\n" };
729 limit $grp 4;
730 feed $grp sub {
731 my $file = pop @files
732 or return;
733
734 add $grp aio_stat $file, sub { ... };
735 };
736
737=item limit $grp $num
738
739Sets the feeder limit for the group: The feeder will be called whenever
740the group contains less than this many requests.
741
742Setting the limit to C<0> will pause the feeding process.
620 743
621=back 744=back
622 745
623=head2 SUPPORT FUNCTIONS 746=head2 SUPPORT FUNCTIONS
624 747
682 IO::AIO::poll_wait, IO::AIO::poll_cb 805 IO::AIO::poll_wait, IO::AIO::poll_cb
683 if IO::AIO::nreqs; 806 if IO::AIO::nreqs;
684 807
685=item IO::AIO::min_parallel $nthreads 808=item IO::AIO::min_parallel $nthreads
686 809
687Set the minimum number of AIO threads to C<$nthreads>. The current default 810Set the minimum number of AIO threads to C<$nthreads>. The current
688is C<4>, which means four asynchronous operations can be done at one time 811default is C<8>, which means eight asynchronous operations can execute
689(the number of outstanding operations, however, is unlimited). 812concurrently at any one time (the number of outstanding requests,
813however, is unlimited).
690 814
691IO::AIO starts threads only on demand, when an AIO request is queued and 815IO::AIO starts threads only on demand, when an AIO request is queued and
692no free thread exists. 816no free thread exists.
693 817
694It is recommended to keep the number of threads low, as some Linux 818It is recommended to keep the number of threads relatively low, as some
695kernel versions will scale negatively with the number of threads (higher 819Linux kernel versions will scale negatively with the number of threads
696parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32 820(higher parallelity => MUCH higher latency). With current Linux 2.6
697threads should be fine. 821versions, 4-32 threads should be fine.
698 822
699Under most circumstances you don't need to call this function, as the 823Under most circumstances you don't need to call this function, as the
700module selects a default that is suitable for low to moderate load. 824module selects a default that is suitable for low to moderate load.
701 825
702=item IO::AIO::max_parallel $nthreads 826=item IO::AIO::max_parallel $nthreads
713 837
714Under normal circumstances you don't need to call this function. 838Under normal circumstances you don't need to call this function.
715 839
716=item $oldnreqs = IO::AIO::max_outstanding $nreqs 840=item $oldnreqs = IO::AIO::max_outstanding $nreqs
717 841
842[DEPRECATED]
843
718Sets the maximum number of outstanding requests to C<$nreqs>. If you 844Sets the maximum number of outstanding requests to C<$nreqs>. If you
719try to queue up more than this number of requests, the caller will block until 845try to queue up more than this number of requests, the caller will block until
720some requests have been handled. 846some requests have been handled.
721 847
722The default is very large, so normally there is no practical limit. If you 848The default is very large, so normally there is no practical limit. If you
723queue up many requests in a loop it often improves speed if you set 849queue up many requests in a loop it often improves speed if you set
724this to a relatively low number, such as C<100>. 850this to a relatively low number, such as C<100>.
851
852This function does not work well together with C<aio_group>'s, and their
853feeder interface is better suited to limiting concurrency, so do not use
854this function.
725 855
726Under normal circumstances you don't need to call this function. 856Under normal circumstances you don't need to call this function.
727 857
728=back 858=back
729 859
743 or return undef; 873 or return undef;
744 874
745 *$sym 875 *$sym
746} 876}
747 877
748min_parallel 4; 878min_parallel 8;
749 879
750END { 880END {
751 max_parallel 0; 881 max_parallel 0;
752} 882}
753 883
767 897
768In short: the parent will, after a short pause, continue as if fork had 898In short: the parent will, after a short pause, continue as if fork had
769not been called, while the child will act as if IO::AIO has not been used 899not been called, while the child will act as if IO::AIO has not been used
770yet. 900yet.
771 901
902=head2 MEMORY USAGE
903
904Each aio request uses - depending on your architecture - around 128 bytes
905of memory. In addition, stat requests need a stat buffer (possibly a few
906hundred bytes). Perl scalars and other data passed into aio requests will
907also be locked.
908
909This is now awfully much, so queuing lots of requests is not usually a
910problem.
911
912Each thread needs a stack area which is usually around 16k, sometimes much
913larger, depending on the OS.
914
772=head1 SEE ALSO 915=head1 SEE ALSO
773 916
774L<Coro>, L<Linux::AIO> (obsolete). 917L<Coro::AIO>.
775 918
776=head1 AUTHOR 919=head1 AUTHOR
777 920
778 Marc Lehmann <schmorp@schmorp.de> 921 Marc Lehmann <schmorp@schmorp.de>
779 http://home.schmorp.de/ 922 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines