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.75 by root, Tue Oct 24 20:10:26 2006 UTC vs.
Revision 1.83 by root, Fri Oct 27 20:11:58 2006 UTC

131 our $VERSION = '2.0'; 131 our $VERSION = '2.0';
132 132
133 our @AIO_REQ = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 133 our @AIO_REQ = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
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_copy aio_group aio_nop aio_mknod);
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):
261 264
262 aio_read $fh, 7, 15, $buffer, 0, sub { 265 aio_read $fh, 7, 15, $buffer, 0, sub {
263 $_[0] > 0 or die "read error: $!"; 266 $_[0] > 0 or die "read error: $!";
264 print "read $_[0] bytes: <$buffer>\n"; 267 print "read $_[0] bytes: <$buffer>\n";
265 }; 268 };
266
267=item aio_move $srcpath, $dstpath, $callback->($status)
268
269Try to move the I<file> (directories not supported as either source or
270destination) from C<$srcpath> to C<$dstpath> and call the callback with
271the C<0> (error) or C<-1> ok.
272
273This is a composite request that tries to rename(2) the file first. If
274rename files with C<EXDEV>, it creates the destination file with mode 0200
275and copies the contents of the source file into it using C<aio_sendfile>,
276followed by restoring atime, mtime, access mode and uid/gid, in that
277order, and unlinking the C<$srcpath>.
278
279If an error occurs, the partial destination file will be unlinked, if
280possible, except when setting atime, mtime, access mode and uid/gid, where
281errors are being ignored.
282
283=cut
284
285sub aio_move($$$) {
286 my ($src, $dst, $cb) = @_;
287
288 my $grp = aio_group $cb;
289
290 add $grp aio_rename $src, $dst, sub {
291 if ($_[0] && $! == EXDEV) {
292 add $grp aio_open $src, O_RDONLY, 0, sub {
293 if (my $src_fh = $_[0]) {
294 my @stat = stat $src_fh;
295
296 add $grp aio_open $dst, O_WRONLY, 0200, sub {
297 if (my $dst_fh = $_[0]) {
298 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
299 close $src_fh;
300
301 if ($_[0] == $stat[7]) {
302 utime $stat[8], $stat[9], $dst;
303 chmod $stat[2] & 07777, $dst_fh;
304 chown $stat[4], $stat[5], $dst_fh;
305 close $dst_fh;
306
307 add $grp aio_unlink $src, sub {
308 $grp->result ($_[0]);
309 };
310 } else {
311 my $errno = $!;
312 add $grp aio_unlink $dst, sub {
313 $! = $errno;
314 $grp->result (-1);
315 };
316 }
317 };
318 } else {
319 $grp->result (-1);
320 }
321 },
322
323 } else {
324 $grp->result (-1);
325 }
326 };
327 } else {
328 $grp->result ($_[0]);
329 }
330 };
331
332 $grp
333}
334 269
335=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval) 270=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
336 271
337Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts 272Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
338reading at byte offset C<$in_offset>, and starts writing at the current 273reading at byte offset C<$in_offset>, and starts writing at the current
394=item aio_unlink $pathname, $callback->($status) 329=item aio_unlink $pathname, $callback->($status)
395 330
396Asynchronously unlink (delete) a file and call the callback with the 331Asynchronously unlink (delete) a file and call the callback with the
397result code. 332result code.
398 333
334=item aio_mknod $path, $mode, $dev, $callback->($status)
335
336Asynchronously create a device node (or fifo). See mknod(2).
337
338The only portable (POSIX) way of calling this function is:
339
340 aio_mknod $path, IO::AIO::S_IFIFO | $mode, 0, sub { ...
341
399=item aio_link $srcpath, $dstpath, $callback->($status) 342=item aio_link $srcpath, $dstpath, $callback->($status)
400 343
401Asynchronously create a new link to the existing object at C<$srcpath> at 344Asynchronously create a new link to the existing object at C<$srcpath> at
402the path C<$dstpath> and call the callback with the result code. 345the path C<$dstpath> and call the callback with the result code.
403 346
423sorted, and will B<NOT> include the C<.> and C<..> entries. 366sorted, and will B<NOT> include the C<.> and C<..> entries.
424 367
425The callback a single argument which is either C<undef> or an array-ref 368The callback a single argument which is either C<undef> or an array-ref
426with the filenames. 369with the filenames.
427 370
371=item aio_copy $srcpath, $dstpath, $callback->($status)
372
373Try to copy the I<file> (directories not supported as either source or
374destination) from C<$srcpath> to C<$dstpath> and call the callback with
375the C<0> (error) or C<-1> ok.
376
377This is a composite request that it creates the destination file with
378mode 0200 and copies the contents of the source file into it using
379C<aio_sendfile>, followed by restoring atime, mtime, access mode and
380uid/gid, in that order.
381
382If an error occurs, the partial destination file will be unlinked, if
383possible, except when setting atime, mtime, access mode and uid/gid, where
384errors are being ignored.
385
386=cut
387
388sub aio_copy($$;$) {
389 my ($src, $dst, $cb) = @_;
390
391 my $pri = aioreq_pri;
392 my $grp = aio_group $cb;
393
394 aioreq_pri $pri;
395 add $grp aio_open $src, O_RDONLY, 0, sub {
396 if (my $src_fh = $_[0]) {
397 my @stat = stat $src_fh;
398
399 aioreq_pri $pri;
400 add $grp aio_open $dst, O_CREAT | O_WRONLY | O_TRUNC, 0200, sub {
401 if (my $dst_fh = $_[0]) {
402 aioreq_pri $pri;
403 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
404 if ($_[0] == $stat[7]) {
405 $grp->result (0);
406 close $src_fh;
407
408 # those should not normally block. should. should.
409 utime $stat[8], $stat[9], $dst;
410 chmod $stat[2] & 07777, $dst_fh;
411 chown $stat[4], $stat[5], $dst_fh;
412 close $dst_fh;
413 } else {
414 $grp->result (-1);
415 close $src_fh;
416 close $dst_fh;
417
418 aioreq $pri;
419 add $grp aio_unlink $dst;
420 }
421 };
422 } else {
423 $grp->result (-1);
424 }
425 },
426
427 } else {
428 $grp->result (-1);
429 }
430 };
431
432 $grp
433}
434
435=item aio_move $srcpath, $dstpath, $callback->($status)
436
437Try to move the I<file> (directories not supported as either source or
438destination) from C<$srcpath> to C<$dstpath> and call the callback with
439the C<0> (error) or C<-1> ok.
440
441This is a composite request that tries to rename(2) the file first. If
442rename files with C<EXDEV>, it copies the file with C<aio_copy> and, if
443that is successful, unlinking the C<$srcpath>.
444
445=cut
446
447sub aio_move($$;$) {
448 my ($src, $dst, $cb) = @_;
449
450 my $pri = aioreq_pri;
451 my $grp = aio_group $cb;
452
453 aioreq_pri $pri;
454 add $grp aio_rename $src, $dst, sub {
455 if ($_[0] && $! == EXDEV) {
456 aioreq_pri $pri;
457 add $grp aio_copy $src, $dst, sub {
458 $grp->result ($_[0]);
459
460 if (!$_[0]) {
461 aioreq_pri $pri;
462 add $grp aio_unlink $src;
463 }
464 };
465 } else {
466 $grp->result ($_[0]);
467 }
468 };
469
470 $grp
471}
472
428=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 473=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
429 474
430Scans a directory (similar to C<aio_readdir>) but additionally tries to 475Scans a directory (similar to C<aio_readdir>) but additionally tries to
431separate the entries of directory C<$path> into two sets of names, ones 476efficiently separate the entries of directory C<$path> into two sets of
432you can recurse into (directories or links to them), and ones you cannot 477names, directories you can recurse into (directories), and ones you cannot
433recurse into (everything else). 478recurse into (everything else, including symlinks to directories).
434 479
435C<aio_scandir> is a composite request that creates of many sub requests_ 480C<aio_scandir> is a composite request that creates of many sub requests_
436C<$maxreq> specifies the maximum number of outstanding aio requests that 481C<$maxreq> specifies the maximum number of outstanding aio requests that
437this function generates. If it is C<< <= 0 >>, then a suitable default 482this function generates. If it is C<< <= 0 >>, then a suitable default
438will be chosen (currently 6). 483will be chosen (currently 4).
439 484
440On error, the callback is called without arguments, otherwise it receives 485On error, the callback is called without arguments, otherwise it receives
441two array-refs with path-relative entry names. 486two array-refs with path-relative entry names.
442 487
443Example: 488Example:
480=cut 525=cut
481 526
482sub aio_scandir($$$) { 527sub aio_scandir($$$) {
483 my ($path, $maxreq, $cb) = @_; 528 my ($path, $maxreq, $cb) = @_;
484 529
530 my $pri = aioreq_pri;
531
485 my $grp = aio_group $cb; 532 my $grp = aio_group $cb;
486 533
487 $maxreq = 6 if $maxreq <= 0; 534 $maxreq = 4 if $maxreq <= 0;
488 535
489 # stat once 536 # stat once
537 aioreq_pri $pri;
490 add $grp aio_stat $path, sub { 538 add $grp aio_stat $path, sub {
491 return $grp->result () if $_[0]; 539 return $grp->result () if $_[0];
492 my $now = time; 540 my $now = time;
493 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 541 my $hash1 = join ":", (stat _)[0,1,3,7,9];
494 542
495 # read the directory entries 543 # read the directory entries
544 aioreq_pri $pri;
496 add $grp aio_readdir $path, sub { 545 add $grp aio_readdir $path, sub {
497 my $entries = shift 546 my $entries = shift
498 or return $grp->result (); 547 or return $grp->result ();
499 548
500 # stat the dir another time 549 # stat the dir another time
550 aioreq_pri $pri;
501 add $grp aio_stat $path, sub { 551 add $grp aio_stat $path, sub {
502 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 552 my $hash2 = join ":", (stat _)[0,1,3,7,9];
503 553
504 my $ndirs; 554 my $ndirs;
505 555
529 limit $statgrp $maxreq; 579 limit $statgrp $maxreq;
530 feed $statgrp sub { 580 feed $statgrp sub {
531 return unless @$entries; 581 return unless @$entries;
532 my $entry = pop @$entries; 582 my $entry = pop @$entries;
533 583
584 aioreq_pri $pri;
534 add $statgrp aio_stat "$path/$entry/.", sub { 585 add $statgrp aio_stat "$path/$entry/.", sub {
535 if ($_[0] < 0) { 586 if ($_[0] < 0) {
536 push @nondirs, $entry; 587 push @nondirs, $entry;
537 } else { 588 } else {
538 # need to check for real directory 589 # need to check for real directory
590 aioreq_pri $pri;
539 add $statgrp aio_lstat "$path/$entry", sub { 591 add $statgrp aio_lstat "$path/$entry", sub {
540 if (-d _) { 592 if (-d _) {
541 push @dirs, $entry; 593 push @dirs, $entry;
542 594
543 unless (--$ndirs) { 595 unless (--$ndirs) {
682=item * They can also can also be added to other IO::AIO::GRP objects. 734=item * They can also can also be added to other IO::AIO::GRP objects.
683 735
684=item * You must not add requests to a group from within the group callback (or 736=item * You must not add requests to a group from within the group callback (or
685any later time). 737any later time).
686 738
687=item * This does not harmonise well with C<max_outstanding>, so best do
688not combine C<aio_group> with it. Groups and feeders are recommended for
689this kind of concurrency-limiting.
690
691=back 739=back
692 740
693Their lifetime, simplified, looks like this: when they are empty, they 741Their lifetime, simplified, looks like this: when they are empty, they
694will finish very quickly. If they contain only requests that are in the 742will finish very quickly. If they contain only requests that are in the
695C<done> state, they will also finish. Otherwise they will continue to 743C<done> state, they will also finish. Otherwise they will continue to
718itself. Useful when you queued a lot of events but got a result early. 766itself. Useful when you queued a lot of events but got a result early.
719 767
720=item $grp->result (...) 768=item $grp->result (...)
721 769
722Set the result value(s) that will be passed to the group callback when all 770Set the result value(s) that will be passed to the group callback when all
723subrequests have finished. By default, no argument will be passed. 771subrequests have finished and set thre groups errno to the current value
772of errno (just like calling C<errno> without an error number). By default,
773no argument will be passed and errno is zero.
774
775=item $grp->errno ([$errno])
776
777Sets the group errno value to C<$errno>, or the current value of errno
778when the argument is missing.
779
780Every aio request has an associated errno value that is restored when
781the callback is invoked. This method lets you change this value from its
782default (0).
783
784Calling C<result> will also set errno, so make sure you either set C<$!>
785before the call to C<result>, or call c<errno> after it.
724 786
725=item feed $grp $callback->($grp) 787=item feed $grp $callback->($grp)
726 788
727Sets a feeder/generator on this group: every group can have an attached 789Sets a feeder/generator on this group: every group can have an attached
728generator that generates requests if idle. The idea behind this is that, 790generator that generates requests if idle. The idea behind this is that,
784 846
785Process all outstanding events on the result pipe. You have to call this 847Process all outstanding events on the result pipe. You have to call this
786regularly. Returns the number of events processed. Returns immediately 848regularly. Returns the number of events processed. Returns immediately
787when no events are outstanding. 849when no events are outstanding.
788 850
851If not all requests were processed for whatever reason, the filehandle
852will still be ready when C<poll_cb> returns.
853
789Example: Install an Event watcher that automatically calls 854Example: Install an Event watcher that automatically calls
790IO::AIO::poll_cb with high priority: 855IO::AIO::poll_cb with high priority:
791 856
792 Event->io (fd => IO::AIO::poll_fileno, 857 Event->io (fd => IO::AIO::poll_fileno,
793 poll => 'r', async => 1, 858 poll => 'r', async => 1,
794 cb => \&IO::AIO::poll_cb); 859 cb => \&IO::AIO::poll_cb);
795 860
861=item IO::AIO::poll_some $max_requests
862
863Similar to C<poll_cb>, but only processes up to C<$max_requests> requests
864at a time.
865
866Useful if you want to ensure some level of interactiveness when perl is
867not fast enough to process all requests in time.
868
869Example: Install an Event watcher that automatically calls
870IO::AIO::poll_some with low priority, to ensure that other parts of the
871program get the CPU sometimes even under high AIO load.
872
873 Event->io (fd => IO::AIO::poll_fileno,
874 poll => 'r', nice => 1,
875 cb => sub { IO::AIO::poll_some 256 });
876
796=item IO::AIO::poll_wait 877=item IO::AIO::poll_wait
797 878
798Wait till the result filehandle becomes ready for reading (simply does a 879Wait till the result filehandle becomes ready for reading (simply does a
799C<select> on the filehandle. This is useful if you want to synchronously wait 880C<select> on the filehandle. This is useful if you want to synchronously wait
800for some requests to finish). 881for some requests to finish).
801 882
802See C<nreqs> for an example. 883See C<nreqs> for an example.
803 884
804=item IO::AIO::nreqs 885=item IO::AIO::nreqs
805 886
806Returns the number of requests currently outstanding (i.e. for which their 887Returns the number of requests currently in the ready, execute or pending
807callback has not been invoked yet). 888states (i.e. for which their callback has not been invoked yet).
808 889
809Example: wait till there are no outstanding requests anymore: 890Example: wait till there are no outstanding requests anymore:
810 891
811 IO::AIO::poll_wait, IO::AIO::poll_cb 892 IO::AIO::poll_wait, IO::AIO::poll_cb
812 while IO::AIO::nreqs; 893 while IO::AIO::nreqs;
894
895=item IO::AIO::nready
896
897Returns the number of requests currently in the ready state (not yet
898executed).
899
900=item IO::AIO::npending
901
902Returns the number of requests currently in the pending state (executed,
903but not yet processed by poll_cb).
813 904
814=item IO::AIO::flush 905=item IO::AIO::flush
815 906
816Wait till all outstanding AIO requests have been handled. 907Wait till all outstanding AIO requests have been handled.
817 908
859This module automatically runs C<max_parallel 0> at program end, to ensure 950This module automatically runs C<max_parallel 0> at program end, to ensure
860that all threads are killed and that there are no outstanding requests. 951that all threads are killed and that there are no outstanding requests.
861 952
862Under normal circumstances you don't need to call this function. 953Under normal circumstances you don't need to call this function.
863 954
864=item $oldnreqs = IO::AIO::max_outstanding $nreqs 955=item $oldmaxreqs = IO::AIO::max_outstanding $maxreqs
865 956
866[DEPRECATED] 957This is a very bad function to use in interactive programs because it
958blocks, and a bad way to reduce concurrency because it is inexact: Better
959use an C<aio_group> together with a feed callback.
867 960
868Sets the maximum number of outstanding requests to C<$nreqs>. If you 961Sets the maximum number of outstanding requests to C<$nreqs>. If you
869try to queue up more than this number of requests, the caller will block until 962to queue up more than this number of requests, the next call to the
870some requests have been handled. 963C<poll_cb> (and C<poll_some> and other functions calling C<poll_cb>)
964function will block until the limit is no longer exceeded.
871 965
872The default is very large, so normally there is no practical limit. If you 966The default value is very large, so there is no practical limit on the
873queue up many requests in a loop it often improves speed if you set 967number of outstanding requests.
874this to a relatively low number, such as C<100>.
875 968
876This function does not work well together with C<aio_group>'s, and their 969You can still queue as many requests as you want. Therefore,
877feeder interface is better suited to limiting concurrency, so do not use 970C<max_oustsanding> is mainly useful in simple scripts (with low values) or
878this function. 971as a stop gap to shield against fatal memory overflow (with large values).
879
880Under normal circumstances you don't need to call this function.
881 972
882=back 973=back
883 974
884=cut 975=cut
885 976
900} 991}
901 992
902min_parallel 8; 993min_parallel 8;
903 994
904END { 995END {
905 max_parallel 0; 996 flush;
906} 997};
907 998
9081; 9991;
909 1000
910=head2 FORK BEHAVIOUR 1001=head2 FORK BEHAVIOUR
911 1002

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines