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.80 by root, Thu Oct 26 16:28:33 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 nready npending); 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';
… …
264 280
265 aio_read $fh, 7, 15, $buffer, 0, sub { 281 aio_read $fh, 7, 15, $buffer, 0, sub {
266 $_[0] > 0 or die "read error: $!"; 282 $_[0] > 0 or die "read error: $!";
267 print "read $_[0] bytes: <$buffer>\n"; 283 print "read $_[0] bytes: <$buffer>\n";
268 }; 284 };
269
270=item aio_move $srcpath, $dstpath, $callback->($status)
271
272Try to move the I<file> (directories not supported as either source or
273destination) from C<$srcpath> to C<$dstpath> and call the callback with
274the C<0> (error) or C<-1> ok.
275
276This is a composite request that tries to rename(2) the file first. If
277rename files with C<EXDEV>, it creates the destination file with mode 0200
278and copies the contents of the source file into it using C<aio_sendfile>,
279followed by restoring atime, mtime, access mode and uid/gid, in that
280order, and unlinking the C<$srcpath>.
281
282If an error occurs, the partial destination file will be unlinked, if
283possible, except when setting atime, mtime, access mode and uid/gid, where
284errors are being ignored.
285
286=cut
287
288sub aio_move($$$) {
289 my ($src, $dst, $cb) = @_;
290
291 my $pri = aioreq_pri;
292 my $grp = aio_group $cb;
293
294 aioreq_pri $pri;
295 add $grp aio_rename $src, $dst, sub {
296 if ($_[0] && $! == EXDEV) {
297 aioreq_pri $pri;
298 add $grp aio_open $src, O_RDONLY, 0, sub {
299 if (my $src_fh = $_[0]) {
300 my @stat = stat $src_fh;
301
302 aioreq_pri $pri;
303 add $grp aio_open $dst, O_WRONLY, 0200, sub {
304 if (my $dst_fh = $_[0]) {
305 aioreq_pri $pri;
306 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
307 close $src_fh;
308
309 if ($_[0] == $stat[7]) {
310 utime $stat[8], $stat[9], $dst;
311 chmod $stat[2] & 07777, $dst_fh;
312 chown $stat[4], $stat[5], $dst_fh;
313 close $dst_fh;
314
315 aioreq_pri $pri;
316 add $grp aio_unlink $src, sub {
317 $grp->result ($_[0]);
318 };
319 } else {
320 my $errno = $!;
321 aioreq_pri $pri;
322 add $grp aio_unlink $dst, sub {
323 $! = $errno;
324 $grp->result (-1);
325 };
326 }
327 };
328 } else {
329 $grp->result (-1);
330 }
331 },
332
333 } else {
334 $grp->result (-1);
335 }
336 };
337 } else {
338 $grp->result ($_[0]);
339 }
340 };
341
342 $grp
343}
344 285
345=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)
346 287
347Tries 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
348reading 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
… …
404=item aio_unlink $pathname, $callback->($status) 345=item aio_unlink $pathname, $callback->($status)
405 346
406Asynchronously unlink (delete) a file and call the callback with the 347Asynchronously unlink (delete) a file and call the callback with the
407result code. 348result code.
408 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
409=item aio_link $srcpath, $dstpath, $callback->($status) 358=item aio_link $srcpath, $dstpath, $callback->($status)
410 359
411Asynchronously 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
412the path C<$dstpath> and call the callback with the result code. 361the path C<$dstpath> and call the callback with the result code.
413 362
… …
432directory (i.e. opendir + readdir + closedir). The entries will not be 381directory (i.e. opendir + readdir + closedir). The entries will not be
433sorted, and will B<NOT> include the C<.> and C<..> entries. 382sorted, and will B<NOT> include the C<.> and C<..> entries.
434 383
435The 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
436with 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}
437 488
438=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 489=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
439 490
440Scans a directory (similar to C<aio_readdir>) but additionally tries to 491Scans a directory (similar to C<aio_readdir>) but additionally tries to
441efficiently separate the entries of directory C<$path> into two sets of 492efficiently separate the entries of directory C<$path> into two sets of
… …
443recurse into (everything else, including symlinks to directories). 494recurse into (everything else, including symlinks to directories).
444 495
445C<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_
446C<$maxreq> specifies the maximum number of outstanding aio requests that 497C<$maxreq> specifies the maximum number of outstanding aio requests that
447this function generates. If it is C<< <= 0 >>, then a suitable default 498this function generates. If it is C<< <= 0 >>, then a suitable default
448will be chosen (currently 6). 499will be chosen (currently 4).
449 500
450On error, the callback is called without arguments, otherwise it receives 501On error, the callback is called without arguments, otherwise it receives
451two array-refs with path-relative entry names. 502two array-refs with path-relative entry names.
452 503
453Example: 504Example:
… …
494 545
495 my $pri = aioreq_pri; 546 my $pri = aioreq_pri;
496 547
497 my $grp = aio_group $cb; 548 my $grp = aio_group $cb;
498 549
499 $maxreq = 6 if $maxreq <= 0; 550 $maxreq = 4 if $maxreq <= 0;
500 551
501 # stat once 552 # stat once
502 aioreq_pri $pri; 553 aioreq_pri $pri;
503 add $grp aio_stat $path, sub { 554 add $grp aio_stat $path, sub {
504 return $grp->result () if $_[0]; 555 return $grp->result () if $_[0];
… …
956} 1007}
957 1008
958min_parallel 8; 1009min_parallel 8;
959 1010
960END { 1011END {
961 max_parallel 0; 1012 min_parallel 1;
962} 1013 flush;
1014};
963 1015
9641; 10161;
965 1017
966=head2 FORK BEHAVIOUR 1018=head2 FORK BEHAVIOUR
967 1019

Diff Legend

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