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.73 by root, Tue Oct 24 16:35:04 2006 UTC vs.
Revision 1.81 by root, Fri Oct 27 19:17:23 2006 UTC

134 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink 134 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
135 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move 135 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
136 aio_group aio_nop); 136 aio_group aio_nop);
137 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice)); 137 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice));
138 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush 138 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush
139 min_parallel max_parallel max_outstanding nreqs); 139 min_parallel max_parallel nreqs nready npending);
140 140
141 @IO::AIO::GRP::ISA = 'IO::AIO::REQ'; 141 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
142 142
143 require XSLoader; 143 require XSLoader;
144 XSLoader::load ("IO::AIO", $VERSION); 144 XSLoader::load ("IO::AIO", $VERSION);
175environment, d) use Glib::filename_from_unicode on unicode filenames or e) 175environment, d) use Glib::filename_from_unicode on unicode filenames or e)
176use something else. 176use something else.
177 177
178=over 4 178=over 4
179 179
180=item aioreq_pri $pri 180=item $prev_pri = aioreq_pri [$pri]
181 181
182Sets the priority for the next aio request. The default priority 182Returns the priority value that would be used for the next request and, if
183C<$pri> is given, sets the priority for the next aio request.
184
183is C<0>, the minimum and maximum priorities are C<-4> and C<4>, 185The default priority is C<0>, the minimum and maximum priorities are C<-4>
184respectively. Requests with higher priority will be serviced first. 186and C<4>, respectively. Requests with higher priority will be serviced
187first.
185 188
186The priority will be reset to C<0> after each call to one of the C<aio_> 189The priority will be reset to C<0> after each call to one of the C<aio_*>
187functions. 190functions.
188 191
189Example: open a file with low priority, then read something from it with 192Example: open a file with low priority, then read something from it with
190higher priority so the read request is serviced before other low priority 193higher priority so the read request is serviced before other low priority
191open requests (potentially spamming the cache): 194open requests (potentially spamming the cache):
283=cut 286=cut
284 287
285sub aio_move($$$) { 288sub aio_move($$$) {
286 my ($src, $dst, $cb) = @_; 289 my ($src, $dst, $cb) = @_;
287 290
291 my $pri = aioreq_pri;
288 my $grp = aio_group $cb; 292 my $grp = aio_group $cb;
289 293
294 aioreq_pri $pri;
290 add $grp aio_rename $src, $dst, sub { 295 add $grp aio_rename $src, $dst, sub {
291 if ($_[0] && $! == EXDEV) { 296 if ($_[0] && $! == EXDEV) {
297 aioreq_pri $pri;
292 add $grp aio_open $src, O_RDONLY, 0, sub { 298 add $grp aio_open $src, O_RDONLY, 0, sub {
293 if (my $src_fh = $_[0]) { 299 if (my $src_fh = $_[0]) {
294 my @stat = stat $src_fh; 300 my @stat = stat $src_fh;
295 301
302 aioreq_pri $pri;
296 add $grp aio_open $dst, O_WRONLY, 0200, sub { 303 add $grp aio_open $dst, O_WRONLY, 0200, sub {
297 if (my $dst_fh = $_[0]) { 304 if (my $dst_fh = $_[0]) {
305 aioreq_pri $pri;
298 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub { 306 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
299 close $src_fh; 307 close $src_fh;
300 308
301 if ($_[0] == $stat[7]) { 309 if ($_[0] == $stat[7]) {
302 utime $stat[8], $stat[9], $dst; 310 utime $stat[8], $stat[9], $dst;
303 chmod $stat[2] & 07777, $dst_fh; 311 chmod $stat[2] & 07777, $dst_fh;
304 chown $stat[4], $stat[5], $dst_fh; 312 chown $stat[4], $stat[5], $dst_fh;
305 close $dst_fh; 313 close $dst_fh;
306 314
315 aioreq_pri $pri;
307 add $grp aio_unlink $src, sub { 316 add $grp aio_unlink $src, sub {
308 $grp->result ($_[0]); 317 $grp->result ($_[0]);
309 }; 318 };
310 } else { 319 } else {
311 my $errno = $!; 320 my $errno = $!;
321 aioreq_pri $pri;
312 add $grp aio_unlink $dst, sub { 322 add $grp aio_unlink $dst, sub {
313 $! = $errno; 323 $! = $errno;
314 $grp->result (-1); 324 $grp->result (-1);
315 }; 325 };
316 } 326 }
426with the filenames. 436with the filenames.
427 437
428=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 438=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
429 439
430Scans a directory (similar to C<aio_readdir>) but additionally tries to 440Scans a directory (similar to C<aio_readdir>) but additionally tries to
431separate the entries of directory C<$path> into two sets of names, ones 441efficiently separate the entries of directory C<$path> into two sets of
432you can recurse into (directories or links to them), and ones you cannot 442names, directories you can recurse into (directories), and ones you cannot
433recurse into (everything else). 443recurse into (everything else, including symlinks to directories).
434 444
435C<aio_scandir> is a composite request that creates of many sub requests_ 445C<aio_scandir> is a composite request that creates of many sub requests_
436C<$maxreq> specifies the maximum number of outstanding aio requests that 446C<$maxreq> specifies the maximum number of outstanding aio requests that
437this function generates. If it is C<< <= 0 >>, then a suitable default 447this function generates. If it is C<< <= 0 >>, then a suitable default
438will be chosen (currently 6). 448will be chosen (currently 4).
439 449
440On error, the callback is called without arguments, otherwise it receives 450On error, the callback is called without arguments, otherwise it receives
441two array-refs with path-relative entry names. 451two array-refs with path-relative entry names.
442 452
443Example: 453Example:
480=cut 490=cut
481 491
482sub aio_scandir($$$) { 492sub aio_scandir($$$) {
483 my ($path, $maxreq, $cb) = @_; 493 my ($path, $maxreq, $cb) = @_;
484 494
495 my $pri = aioreq_pri;
496
485 my $grp = aio_group $cb; 497 my $grp = aio_group $cb;
486 498
487 $maxreq = 6 if $maxreq <= 0; 499 $maxreq = 4 if $maxreq <= 0;
488 500
489 # stat once 501 # stat once
502 aioreq_pri $pri;
490 add $grp aio_stat $path, sub { 503 add $grp aio_stat $path, sub {
491 return $grp->result () if $_[0]; 504 return $grp->result () if $_[0];
492 my $now = time; 505 my $now = time;
493 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 506 my $hash1 = join ":", (stat _)[0,1,3,7,9];
494 507
495 # read the directory entries 508 # read the directory entries
509 aioreq_pri $pri;
496 add $grp aio_readdir $path, sub { 510 add $grp aio_readdir $path, sub {
497 my $entries = shift 511 my $entries = shift
498 or return $grp->result (); 512 or return $grp->result ();
499 513
500 # stat the dir another time 514 # stat the dir another time
515 aioreq_pri $pri;
501 add $grp aio_stat $path, sub { 516 add $grp aio_stat $path, sub {
502 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 517 my $hash2 = join ":", (stat _)[0,1,3,7,9];
503 518
504 my $ndirs; 519 my $ndirs;
505 520
520 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length], 535 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
521 @$entries]; 536 @$entries];
522 537
523 my (@dirs, @nondirs); 538 my (@dirs, @nondirs);
524 539
525 my ($statcb, $schedcb);
526 my $nreq = 0;
527
528 my $statgrp = add $grp aio_group; 540 my $statgrp = add $grp aio_group sub {
541 $grp->result (\@dirs, \@nondirs);
542 };
529 543
530 $schedcb = sub { 544 limit $statgrp $maxreq;
531 if (@$entries) { 545 feed $statgrp sub {
532 if ($nreq < $maxreq) { 546 return unless @$entries;
533 my $ent = pop @$entries; 547 my $entry = pop @$entries;
548
549 aioreq_pri $pri;
550 add $statgrp aio_stat "$path/$entry/.", sub {
551 if ($_[0] < 0) {
552 push @nondirs, $entry;
553 } else {
554 # need to check for real directory
555 aioreq_pri $pri;
556 add $statgrp aio_lstat "$path/$entry", sub {
557 if (-d _) {
558 push @dirs, $entry;
559
560 unless (--$ndirs) {
561 push @nondirs, @$entries;
562 feed $statgrp;
563 }
564 } else {
565 push @nondirs, $entry;
566 }
534 $nreq++; 567 }
535 add $statgrp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
536 } 568 }
537 } elsif (!$nreq) {
538 # finished
539 $statgrp->cancel;
540 undef $statcb;
541 undef $schedcb;
542 $grp->result (\@dirs, \@nondirs);
543 } 569 };
544 }; 570 };
545 $statcb = sub {
546 my ($status, $entry) = @_;
547
548 if ($status < 0) {
549 $nreq--;
550 push @nondirs, $entry;
551 &$schedcb;
552 } else {
553 # need to check for real directory
554 add $grp aio_lstat "$path/$entry", sub {
555 $nreq--;
556
557 if (-d _) {
558 push @dirs, $entry;
559
560 if (!--$ndirs) {
561 push @nondirs, @$entries;
562 $entries = [];
563 }
564 } else {
565 push @nondirs, $entry;
566 }
567
568 &$schedcb;
569 }
570 }
571 };
572
573 &$schedcb while @$entries && $nreq < $maxreq;
574 }; 571 };
575 }; 572 };
576 }; 573 };
577 574
578 $grp 575 $grp
590 587
591If this call isn't available because your OS lacks it or it couldn't be 588If this call isn't available because your OS lacks it or it couldn't be
592detected, it will be emulated by calling C<fsync> instead. 589detected, it will be emulated by calling C<fsync> instead.
593 590
594=item aio_group $callback->(...) 591=item aio_group $callback->(...)
595
596[EXPERIMENTAL]
597 592
598This is a very special aio request: Instead of doing something, it is a 593This is a very special aio request: Instead of doing something, it is a
599container for other aio requests, which is useful if you want to bundle 594container for other aio requests, which is useful if you want to bundle
600many requests into a single, composite, request with a definite callback 595many requests into a single, composite, request with a definite callback
601and the ability to cancel the whole request with its subrequests. 596and the ability to cancel the whole request with its subrequests.
704=item * They can also can also be added to other IO::AIO::GRP objects. 699=item * They can also can also be added to other IO::AIO::GRP objects.
705 700
706=item * You must not add requests to a group from within the group callback (or 701=item * You must not add requests to a group from within the group callback (or
707any later time). 702any later time).
708 703
709=item * This does not harmonise well with C<max_outstanding>, so best do
710not combine C<aio_group> with it. Groups and feeders are recommended for
711this kind of concurrency-limiting.
712
713=back 704=back
714 705
715Their lifetime, simplified, looks like this: when they are empty, they 706Their lifetime, simplified, looks like this: when they are empty, they
716will finish very quickly. If they contain only requests that are in the 707will finish very quickly. If they contain only requests that are in the
717C<done> state, they will also finish. Otherwise they will continue to 708C<done> state, they will also finish. Otherwise they will continue to
732be added, including other groups, as long as you do not create circular 723be added, including other groups, as long as you do not create circular
733dependencies. 724dependencies.
734 725
735Returns all its arguments. 726Returns all its arguments.
736 727
728=item $grp->cancel_subs
729
730Cancel all subrequests and clears any feeder, but not the group request
731itself. Useful when you queued a lot of events but got a result early.
732
737=item $grp->result (...) 733=item $grp->result (...)
738 734
739Set the result value(s) that will be passed to the group callback when all 735Set the result value(s) that will be passed to the group callback when all
740subrequests have finished. By default, no argument will be passed. 736subrequests have finished and set thre groups errno to the current value
737of errno (just like calling C<errno> without an error number). By default,
738no argument will be passed and errno is zero.
739
740=item $grp->errno ([$errno])
741
742Sets the group errno value to C<$errno>, or the current value of errno
743when the argument is missing.
744
745Every aio request has an associated errno value that is restored when
746the callback is invoked. This method lets you change this value from its
747default (0).
748
749Calling C<result> will also set errno, so make sure you either set C<$!>
750before the call to C<result>, or call c<errno> after it.
741 751
742=item feed $grp $callback->($grp) 752=item feed $grp $callback->($grp)
743
744[VERY EXPERIMENTAL]
745 753
746Sets a feeder/generator on this group: every group can have an attached 754Sets a feeder/generator on this group: every group can have an attached
747generator that generates requests if idle. The idea behind this is that, 755generator that generates requests if idle. The idea behind this is that,
748although you could just queue as many requests as you want in a group, 756although you could just queue as many requests as you want in a group,
749this might starve other requests for a potentially long time. For 757this might starve other requests for a potentially long time. For
803 811
804Process all outstanding events on the result pipe. You have to call this 812Process all outstanding events on the result pipe. You have to call this
805regularly. Returns the number of events processed. Returns immediately 813regularly. Returns the number of events processed. Returns immediately
806when no events are outstanding. 814when no events are outstanding.
807 815
816If not all requests were processed for whatever reason, the filehandle
817will still be ready when C<poll_cb> returns.
818
808Example: Install an Event watcher that automatically calls 819Example: Install an Event watcher that automatically calls
809IO::AIO::poll_cb with high priority: 820IO::AIO::poll_cb with high priority:
810 821
811 Event->io (fd => IO::AIO::poll_fileno, 822 Event->io (fd => IO::AIO::poll_fileno,
812 poll => 'r', async => 1, 823 poll => 'r', async => 1,
813 cb => \&IO::AIO::poll_cb); 824 cb => \&IO::AIO::poll_cb);
814 825
826=item IO::AIO::poll_some $max_requests
827
828Similar to C<poll_cb>, but only processes up to C<$max_requests> requests
829at a time.
830
831Useful if you want to ensure some level of interactiveness when perl is
832not fast enough to process all requests in time.
833
834Example: Install an Event watcher that automatically calls
835IO::AIO::poll_some with low priority, to ensure that other parts of the
836program get the CPU sometimes even under high AIO load.
837
838 Event->io (fd => IO::AIO::poll_fileno,
839 poll => 'r', nice => 1,
840 cb => sub { IO::AIO::poll_some 256 });
841
815=item IO::AIO::poll_wait 842=item IO::AIO::poll_wait
816 843
817Wait till the result filehandle becomes ready for reading (simply does a 844Wait till the result filehandle becomes ready for reading (simply does a
818C<select> on the filehandle. This is useful if you want to synchronously wait 845C<select> on the filehandle. This is useful if you want to synchronously wait
819for some requests to finish). 846for some requests to finish).
820 847
821See C<nreqs> for an example. 848See C<nreqs> for an example.
822 849
823=item IO::AIO::nreqs 850=item IO::AIO::nreqs
824 851
825Returns the number of requests currently outstanding (i.e. for which their 852Returns the number of requests currently in the ready, execute or pending
826callback has not been invoked yet). 853states (i.e. for which their callback has not been invoked yet).
827 854
828Example: wait till there are no outstanding requests anymore: 855Example: wait till there are no outstanding requests anymore:
829 856
830 IO::AIO::poll_wait, IO::AIO::poll_cb 857 IO::AIO::poll_wait, IO::AIO::poll_cb
831 while IO::AIO::nreqs; 858 while IO::AIO::nreqs;
859
860=item IO::AIO::nready
861
862Returns the number of requests currently in the ready state (not yet
863executed).
864
865=item IO::AIO::npending
866
867Returns the number of requests currently in the pending state (executed,
868but not yet processed by poll_cb).
832 869
833=item IO::AIO::flush 870=item IO::AIO::flush
834 871
835Wait till all outstanding AIO requests have been handled. 872Wait till all outstanding AIO requests have been handled.
836 873
878This module automatically runs C<max_parallel 0> at program end, to ensure 915This module automatically runs C<max_parallel 0> at program end, to ensure
879that all threads are killed and that there are no outstanding requests. 916that all threads are killed and that there are no outstanding requests.
880 917
881Under normal circumstances you don't need to call this function. 918Under normal circumstances you don't need to call this function.
882 919
883=item $oldnreqs = IO::AIO::max_outstanding $nreqs 920=item $oldmaxreqs = IO::AIO::max_outstanding $maxreqs
884 921
885[DEPRECATED] 922This is a very bad function to use in interactive programs because it
923blocks, and a bad way to reduce concurrency because it is inexact: Better
924use an C<aio_group> together with a feed callback.
886 925
887Sets the maximum number of outstanding requests to C<$nreqs>. If you 926Sets the maximum number of outstanding requests to C<$nreqs>. If you
888try to queue up more than this number of requests, the caller will block until 927to queue up more than this number of requests, the next call to the
889some requests have been handled. 928C<poll_cb> (and C<poll_some> and other functions calling C<poll_cb>)
929function will block until the limit is no longer exceeded.
890 930
891The default is very large, so normally there is no practical limit. If you 931The default value is very large, so there is no practical limit on the
892queue up many requests in a loop it often improves speed if you set 932number of outstanding requests.
893this to a relatively low number, such as C<100>.
894 933
895This function does not work well together with C<aio_group>'s, and their 934You can still queue as many requests as you want. Therefore,
896feeder interface is better suited to limiting concurrency, so do not use 935C<max_oustsanding> is mainly useful in simple scripts (with low values) or
897this function. 936as a stop gap to shield against fatal memory overflow (with large values).
898
899Under normal circumstances you don't need to call this function.
900 937
901=back 938=back
902 939
903=cut 940=cut
904 941
917 954
918 *$sym 955 *$sym
919} 956}
920 957
921min_parallel 8; 958min_parallel 8;
922
923END {
924 max_parallel 0;
925}
926 959
9271; 9601;
928 961
929=head2 FORK BEHAVIOUR 962=head2 FORK BEHAVIOUR
930 963

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines