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.79 by root, Thu Oct 26 14:35:34 2006 UTC vs.
Revision 1.85 by root, Sat Oct 28 01:40:30 2006 UTC

… …
50 50
51=head1 DESCRIPTION 51=head1 DESCRIPTION
52 52
53This module implements asynchronous I/O using whatever means your 53This module implements asynchronous I/O using whatever means your
54operating system supports. 54operating system supports.
55
56Asynchronous means that operations that can normally block your program
57(e.g. reading from disk) will be done asynchronously: the operation
58will still block, but you can do something else in the meantime. This
59is extremely useful for programs that need to stay interactive even
60when doing heavy I/O (GUI programs, high performance network servers
61etc.), but can also be used to easily do operations in parallel that are
62normally done sequentially, e.g. stat'ing many files, which is much faster
63on a RAID volume or over NFS when you do a number of stat operations
64concurrently.
65
66While this works on all types of file descriptors (for example sockets),
67using these functions on file descriptors that support nonblocking
68operation (again, sockets, pipes etc.) is very inefficient. Use an event
69loop for that (such as the L<Event|Event> module): IO::AIO will naturally
70fit into such an event loop itself.
55 71
56In this version, a number of threads are started that execute your 72In this version, a number of threads are started that execute your
57requests and signal their completion. You don't need thread support 73requests and signal their completion. You don't need thread support
58in perl, and the threads created by this module will not be visible 74in perl, and the threads created by this module will not be visible
59to perl. In the future, this module might make use of the native aio 75to perl. In the future, this module might make use of the native aio
60functions available on many operating systems. However, they are often 76functions available on many operating systems. However, they are often
61not well-supported or restricted (Linux doesn't allow them on normal 77not well-supported or restricted (GNU/Linux doesn't allow them on normal
62files currently, for example), and they would only support aio_read and 78files currently, for example), and they would only support aio_read and
63aio_write, so the remaining functionality would have to be implemented 79aio_write, so the remaining functionality would have to be implemented
64using threads anyway. 80using threads anyway.
65 81
66Although the module will work with in the presence of other (Perl-) 82Although the module will work with in the presence of other (Perl-)
… …
131 our $VERSION = '2.0'; 147 our $VERSION = '2.0';
132 148
133 our @AIO_REQ = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 149 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 150 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 151 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
136 aio_group aio_nop); 152 aio_copy aio_group aio_nop aio_mknod);
137 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice)); 153 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice));
138 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush 154 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush
139 min_parallel max_parallel nreqs); 155 min_parallel max_parallel nreqs nready npending);
140 156
141 @IO::AIO::GRP::ISA = 'IO::AIO::REQ'; 157 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
142 158
143 require XSLoader; 159 require XSLoader;
144 XSLoader::load ("IO::AIO", $VERSION); 160 XSLoader::load ("IO::AIO", $VERSION);
… …
175environment, d) use Glib::filename_from_unicode on unicode filenames or e) 191environment, d) use Glib::filename_from_unicode on unicode filenames or e)
176use something else. 192use something else.
177 193
178=over 4 194=over 4
179 195
180=item aioreq_pri $pri 196=item $prev_pri = aioreq_pri [$pri]
181 197
182Sets the priority for the next aio request. The default priority 198Returns the priority value that would be used for the next request and, if
199C<$pri> is given, sets the priority for the next aio request.
200
183is C<0>, the minimum and maximum priorities are C<-4> and C<4>, 201The default priority is C<0>, the minimum and maximum priorities are C<-4>
184respectively. Requests with higher priority will be serviced first. 202and C<4>, respectively. Requests with higher priority will be serviced
203first.
185 204
186The priority will be reset to C<0> after each call to one of the C<aio_> 205The priority will be reset to C<0> after each call to one of the C<aio_*>
187functions. 206functions.
188 207
189Example: open a file with low priority, then read something from it with 208Example: open a file with low priority, then read something from it with
190higher priority so the read request is serviced before other low priority 209higher priority so the read request is serviced before other low priority
191open requests (potentially spamming the cache): 210open requests (potentially spamming the cache):
… …
261 280
262 aio_read $fh, 7, 15, $buffer, 0, sub { 281 aio_read $fh, 7, 15, $buffer, 0, sub {
263 $_[0] > 0 or die "read error: $!"; 282 $_[0] > 0 or die "read error: $!";
264 print "read $_[0] bytes: <$buffer>\n"; 283 print "read $_[0] bytes: <$buffer>\n";
265 }; 284 };
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 285
335=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval) 286=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
336 287
337Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts 288Tries 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 289reading at byte offset C<$in_offset>, and starts writing at the current
… …
394=item aio_unlink $pathname, $callback->($status) 345=item aio_unlink $pathname, $callback->($status)
395 346
396Asynchronously unlink (delete) a file and call the callback with the 347Asynchronously unlink (delete) a file and call the callback with the
397result code. 348result code.
398 349
350=item aio_mknod $path, $mode, $dev, $callback->($status)
351
352Asynchronously create a device node (or fifo). See mknod(2).
353
354The only portable (POSIX) way of calling this function is:
355
356 aio_mknod $path, IO::AIO::S_IFIFO | $mode, 0, sub { ...
357
399=item aio_link $srcpath, $dstpath, $callback->($status) 358=item aio_link $srcpath, $dstpath, $callback->($status)
400 359
401Asynchronously create a new link to the existing object at C<$srcpath> at 360Asynchronously create a new link to the existing object at C<$srcpath> at
402the path C<$dstpath> and call the callback with the result code. 361the path C<$dstpath> and call the callback with the result code.
403 362
… …
422directory (i.e. opendir + readdir + closedir). The entries will not be 381directory (i.e. opendir + readdir + closedir). The entries will not be
423sorted, and will B<NOT> include the C<.> and C<..> entries. 382sorted, and will B<NOT> include the C<.> and C<..> entries.
424 383
425The callback a single argument which is either C<undef> or an array-ref 384The callback a single argument which is either C<undef> or an array-ref
426with the filenames. 385with the filenames.
386
387=item aio_copy $srcpath, $dstpath, $callback->($status)
388
389Try to copy the I<file> (directories not supported as either source or
390destination) from C<$srcpath> to C<$dstpath> and call the callback with
391the C<0> (error) or C<-1> ok.
392
393This is a composite request that it creates the destination file with
394mode 0200 and copies the contents of the source file into it using
395C<aio_sendfile>, followed by restoring atime, mtime, access mode and
396uid/gid, in that order.
397
398If an error occurs, the partial destination file will be unlinked, if
399possible, except when setting atime, mtime, access mode and uid/gid, where
400errors are being ignored.
401
402=cut
403
404sub aio_copy($$;$) {
405 my ($src, $dst, $cb) = @_;
406
407 my $pri = aioreq_pri;
408 my $grp = aio_group $cb;
409
410 aioreq_pri $pri;
411 add $grp aio_open $src, O_RDONLY, 0, sub {
412 if (my $src_fh = $_[0]) {
413 my @stat = stat $src_fh;
414
415 aioreq_pri $pri;
416 add $grp aio_open $dst, O_CREAT | O_WRONLY | O_TRUNC, 0200, sub {
417 if (my $dst_fh = $_[0]) {
418 aioreq_pri $pri;
419 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
420 if ($_[0] == $stat[7]) {
421 $grp->result (0);
422 close $src_fh;
423
424 # those should not normally block. should. should.
425 utime $stat[8], $stat[9], $dst;
426 chmod $stat[2] & 07777, $dst_fh;
427 chown $stat[4], $stat[5], $dst_fh;
428 close $dst_fh;
429 } else {
430 $grp->result (-1);
431 close $src_fh;
432 close $dst_fh;
433
434 aioreq $pri;
435 add $grp aio_unlink $dst;
436 }
437 };
438 } else {
439 $grp->result (-1);
440 }
441 },
442
443 } else {
444 $grp->result (-1);
445 }
446 };
447
448 $grp
449}
450
451=item aio_move $srcpath, $dstpath, $callback->($status)
452
453Try to move the I<file> (directories not supported as either source or
454destination) from C<$srcpath> to C<$dstpath> and call the callback with
455the C<0> (error) or C<-1> ok.
456
457This is a composite request that tries to rename(2) the file first. If
458rename files with C<EXDEV>, it copies the file with C<aio_copy> and, if
459that is successful, unlinking the C<$srcpath>.
460
461=cut
462
463sub aio_move($$;$) {
464 my ($src, $dst, $cb) = @_;
465
466 my $pri = aioreq_pri;
467 my $grp = aio_group $cb;
468
469 aioreq_pri $pri;
470 add $grp aio_rename $src, $dst, sub {
471 if ($_[0] && $! == EXDEV) {
472 aioreq_pri $pri;
473 add $grp aio_copy $src, $dst, sub {
474 $grp->result ($_[0]);
475
476 if (!$_[0]) {
477 aioreq_pri $pri;
478 add $grp aio_unlink $src;
479 }
480 };
481 } else {
482 $grp->result ($_[0]);
483 }
484 };
485
486 $grp
487}
427 488
428=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 489=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
429 490
430Scans a directory (similar to C<aio_readdir>) but additionally tries to 491Scans a directory (similar to C<aio_readdir>) but additionally tries to
431efficiently separate the entries of directory C<$path> into two sets of 492efficiently separate the entries of directory C<$path> into two sets of
… …
433recurse into (everything else, including symlinks to directories). 494recurse into (everything else, including symlinks to directories).
434 495
435C<aio_scandir> is a composite request that creates of many sub requests_ 496C<aio_scandir> is a composite request that creates of many sub requests_
436C<$maxreq> specifies the maximum number of outstanding aio requests that 497C<$maxreq> specifies the maximum number of outstanding aio requests that
437this function generates. If it is C<< <= 0 >>, then a suitable default 498this function generates. If it is C<< <= 0 >>, then a suitable default
438will be chosen (currently 6). 499will be chosen (currently 4).
439 500
440On error, the callback is called without arguments, otherwise it receives 501On error, the callback is called without arguments, otherwise it receives
441two array-refs with path-relative entry names. 502two array-refs with path-relative entry names.
442 503
443Example: 504Example:
… …
480=cut 541=cut
481 542
482sub aio_scandir($$$) { 543sub aio_scandir($$$) {
483 my ($path, $maxreq, $cb) = @_; 544 my ($path, $maxreq, $cb) = @_;
484 545
546 my $pri = aioreq_pri;
547
485 my $grp = aio_group $cb; 548 my $grp = aio_group $cb;
486 549
487 $maxreq = 6 if $maxreq <= 0; 550 $maxreq = 4 if $maxreq <= 0;
488 551
489 # stat once 552 # stat once
553 aioreq_pri $pri;
490 add $grp aio_stat $path, sub { 554 add $grp aio_stat $path, sub {
491 return $grp->result () if $_[0]; 555 return $grp->result () if $_[0];
492 my $now = time; 556 my $now = time;
493 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 557 my $hash1 = join ":", (stat _)[0,1,3,7,9];
494 558
495 # read the directory entries 559 # read the directory entries
560 aioreq_pri $pri;
496 add $grp aio_readdir $path, sub { 561 add $grp aio_readdir $path, sub {
497 my $entries = shift 562 my $entries = shift
498 or return $grp->result (); 563 or return $grp->result ();
499 564
500 # stat the dir another time 565 # stat the dir another time
566 aioreq_pri $pri;
501 add $grp aio_stat $path, sub { 567 add $grp aio_stat $path, sub {
502 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 568 my $hash2 = join ":", (stat _)[0,1,3,7,9];
503 569
504 my $ndirs; 570 my $ndirs;
505 571
… …
529 limit $statgrp $maxreq; 595 limit $statgrp $maxreq;
530 feed $statgrp sub { 596 feed $statgrp sub {
531 return unless @$entries; 597 return unless @$entries;
532 my $entry = pop @$entries; 598 my $entry = pop @$entries;
533 599
600 aioreq_pri $pri;
534 add $statgrp aio_stat "$path/$entry/.", sub { 601 add $statgrp aio_stat "$path/$entry/.", sub {
535 if ($_[0] < 0) { 602 if ($_[0] < 0) {
536 push @nondirs, $entry; 603 push @nondirs, $entry;
537 } else { 604 } else {
538 # need to check for real directory 605 # need to check for real directory
606 aioreq_pri $pri;
539 add $statgrp aio_lstat "$path/$entry", sub { 607 add $statgrp aio_lstat "$path/$entry", sub {
540 if (-d _) { 608 if (-d _) {
541 push @dirs, $entry; 609 push @dirs, $entry;
542 610
543 unless (--$ndirs) { 611 unless (--$ndirs) {
… …
714itself. Useful when you queued a lot of events but got a result early. 782itself. Useful when you queued a lot of events but got a result early.
715 783
716=item $grp->result (...) 784=item $grp->result (...)
717 785
718Set the result value(s) that will be passed to the group callback when all 786Set the result value(s) that will be passed to the group callback when all
719subrequests have finished. By default, no argument will be passed. 787subrequests have finished and set thre groups errno to the current value
788of errno (just like calling C<errno> without an error number). By default,
789no argument will be passed and errno is zero.
790
791=item $grp->errno ([$errno])
792
793Sets the group errno value to C<$errno>, or the current value of errno
794when the argument is missing.
795
796Every aio request has an associated errno value that is restored when
797the callback is invoked. This method lets you change this value from its
798default (0).
799
800Calling C<result> will also set errno, so make sure you either set C<$!>
801before the call to C<result>, or call c<errno> after it.
720 802
721=item feed $grp $callback->($grp) 803=item feed $grp $callback->($grp)
722 804
723Sets a feeder/generator on this group: every group can have an attached 805Sets a feeder/generator on this group: every group can have an attached
724generator that generates requests if idle. The idea behind this is that, 806generator that generates requests if idle. The idea behind this is that,
… …
816 898
817See C<nreqs> for an example. 899See C<nreqs> for an example.
818 900
819=item IO::AIO::nreqs 901=item IO::AIO::nreqs
820 902
821Returns the number of requests currently outstanding (i.e. for which their 903Returns the number of requests currently in the ready, execute or pending
822callback has not been invoked yet). 904states (i.e. for which their callback has not been invoked yet).
823 905
824Example: wait till there are no outstanding requests anymore: 906Example: wait till there are no outstanding requests anymore:
825 907
826 IO::AIO::poll_wait, IO::AIO::poll_cb 908 IO::AIO::poll_wait, IO::AIO::poll_cb
827 while IO::AIO::nreqs; 909 while IO::AIO::nreqs;
910
911=item IO::AIO::nready
912
913Returns the number of requests currently in the ready state (not yet
914executed).
915
916=item IO::AIO::npending
917
918Returns the number of requests currently in the pending state (executed,
919but not yet processed by poll_cb).
828 920
829=item IO::AIO::flush 921=item IO::AIO::flush
830 922
831Wait till all outstanding AIO requests have been handled. 923Wait till all outstanding AIO requests have been handled.
832 924
… …
915} 1007}
916 1008
917min_parallel 8; 1009min_parallel 8;
918 1010
919END { 1011END {
920 max_parallel 0; 1012 min_parallel 1;
921} 1013 flush;
1014};
922 1015
9231; 10161;
924 1017
925=head2 FORK BEHAVIOUR 1018=head2 FORK BEHAVIOUR
926 1019

Diff Legend

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