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.78 by root, Thu Oct 26 12:38:04 2006 UTC vs.
Revision 1.84 by root, Sat Oct 28 00:17:30 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 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
… …
422directory (i.e. opendir + readdir + closedir). The entries will not be 365directory (i.e. opendir + readdir + closedir). The entries will not be
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.
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}
427 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
431efficiently separate the entries of directory C<$path> into two sets of 476efficiently separate the entries of directory C<$path> into two sets of
… …
433recurse into (everything else, including symlinks to directories). 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) {
… …
714itself. 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.
715 767
716=item $grp->result (...) 768=item $grp->result (...)
717 769
718Set 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
719subrequests 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.
720 786
721=item feed $grp $callback->($grp) 787=item feed $grp $callback->($grp)
722 788
723Sets 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
724generator that generates requests if idle. The idea behind this is that, 790generator that generates requests if idle. The idea behind this is that,
… …
816 882
817See C<nreqs> for an example. 883See C<nreqs> for an example.
818 884
819=item IO::AIO::nreqs 885=item IO::AIO::nreqs
820 886
821Returns the number of requests currently outstanding (i.e. for which their 887Returns the number of requests currently in the ready, execute or pending
822callback has not been invoked yet). 888states (i.e. for which their callback has not been invoked yet).
823 889
824Example: wait till there are no outstanding requests anymore: 890Example: wait till there are no outstanding requests anymore:
825 891
826 IO::AIO::poll_wait, IO::AIO::poll_cb 892 IO::AIO::poll_wait, IO::AIO::poll_cb
827 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).
828 904
829=item IO::AIO::flush 905=item IO::AIO::flush
830 906
831Wait till all outstanding AIO requests have been handled. 907Wait till all outstanding AIO requests have been handled.
832 908
… …
874This 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
875that all threads are killed and that there are no outstanding requests. 951that all threads are killed and that there are no outstanding requests.
876 952
877Under normal circumstances you don't need to call this function. 953Under normal circumstances you don't need to call this function.
878 954
879=item $oldnreqs = IO::AIO::max_outstanding $nreqs 955=item $oldmaxreqs = IO::AIO::max_outstanding $maxreqs
880 956
881[REMOVED] 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.
882 960
883Pre-2.x versions used max_outstanding for a crude request queue length limit.
884
885In 2.x+ you are advised to use a group and a feeder to limit
886concurrency. The max_outstanding feature ran very unstable (endless
887recursions causing segfaults, bad interaction with groups etc.) and was
888removed.
889
890I am deeply sorry, but I am still on the hunt for a good limiting interface.
891
892Original description was as follows:
893
894Sets the maximum number of outstanding requests to C<$nreqs>. If you try 961Sets the maximum number of outstanding requests to C<$nreqs>. If you
895to 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
896some 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.
965
966The default value is very large, so there is no practical limit on the
967number of outstanding requests.
968
969You can still queue as many requests as you want. Therefore,
970C<max_oustsanding> is mainly useful in simple scripts (with low values) or
971as a stop gap to shield against fatal memory overflow (with large values).
897 972
898=back 973=back
899 974
900=cut 975=cut
901 976
… …
916} 991}
917 992
918min_parallel 8; 993min_parallel 8;
919 994
920END { 995END {
921 max_parallel 0; 996 min_parallel 1;
922} 997 flush;
998};
923 999
9241; 10001;
925 1001
926=head2 FORK BEHAVIOUR 1002=head2 FORK BEHAVIOUR
927 1003

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines