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.76 by root, Wed Oct 25 08:20:14 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 max_outstanding 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) {
682=item * They can also can also be added to other IO::AIO::GRP objects. 750=item * They can also can also be added to other IO::AIO::GRP objects.
683 751
684=item * You must not add requests to a group from within the group callback (or 752=item * You must not add requests to a group from within the group callback (or
685any later time). 753any later time).
686 754
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 755=back
692 756
693Their lifetime, simplified, looks like this: when they are empty, they 757Their lifetime, simplified, looks like this: when they are empty, they
694will finish very quickly. If they contain only requests that are in the 758will finish very quickly. If they contain only requests that are in the
695C<done> state, they will also finish. Otherwise they will continue to 759C<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. 782itself. Useful when you queued a lot of events but got a result early.
719 783
720=item $grp->result (...) 784=item $grp->result (...)
721 785
722Set 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
723subrequests 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.
724 802
725=item feed $grp $callback->($grp) 803=item feed $grp $callback->($grp)
726 804
727Sets 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
728generator that generates requests if idle. The idea behind this is that, 806generator that generates requests if idle. The idea behind this is that,
784 862
785Process all outstanding events on the result pipe. You have to call this 863Process all outstanding events on the result pipe. You have to call this
786regularly. Returns the number of events processed. Returns immediately 864regularly. Returns the number of events processed. Returns immediately
787when no events are outstanding. 865when no events are outstanding.
788 866
867If not all requests were processed for whatever reason, the filehandle
868will still be ready when C<poll_cb> returns.
869
789Example: Install an Event watcher that automatically calls 870Example: Install an Event watcher that automatically calls
790IO::AIO::poll_cb with high priority: 871IO::AIO::poll_cb with high priority:
791 872
792 Event->io (fd => IO::AIO::poll_fileno, 873 Event->io (fd => IO::AIO::poll_fileno,
793 poll => 'r', async => 1, 874 poll => 'r', async => 1,
794 cb => \&IO::AIO::poll_cb); 875 cb => \&IO::AIO::poll_cb);
795 876
877=item IO::AIO::poll_some $max_requests
878
879Similar to C<poll_cb>, but only processes up to C<$max_requests> requests
880at a time.
881
882Useful if you want to ensure some level of interactiveness when perl is
883not fast enough to process all requests in time.
884
885Example: Install an Event watcher that automatically calls
886IO::AIO::poll_some with low priority, to ensure that other parts of the
887program get the CPU sometimes even under high AIO load.
888
889 Event->io (fd => IO::AIO::poll_fileno,
890 poll => 'r', nice => 1,
891 cb => sub { IO::AIO::poll_some 256 });
892
796=item IO::AIO::poll_wait 893=item IO::AIO::poll_wait
797 894
798Wait till the result filehandle becomes ready for reading (simply does a 895Wait 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 896C<select> on the filehandle. This is useful if you want to synchronously wait
800for some requests to finish). 897for some requests to finish).
801 898
802See C<nreqs> for an example. 899See C<nreqs> for an example.
803 900
804=item IO::AIO::nreqs 901=item IO::AIO::nreqs
805 902
806Returns the number of requests currently outstanding (i.e. for which their 903Returns the number of requests currently in the ready, execute or pending
807callback has not been invoked yet). 904states (i.e. for which their callback has not been invoked yet).
808 905
809Example: wait till there are no outstanding requests anymore: 906Example: wait till there are no outstanding requests anymore:
810 907
811 IO::AIO::poll_wait, IO::AIO::poll_cb 908 IO::AIO::poll_wait, IO::AIO::poll_cb
812 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).
813 920
814=item IO::AIO::flush 921=item IO::AIO::flush
815 922
816Wait till all outstanding AIO requests have been handled. 923Wait till all outstanding AIO requests have been handled.
817 924
859This module automatically runs C<max_parallel 0> at program end, to ensure 966This module automatically runs C<max_parallel 0> at program end, to ensure
860that all threads are killed and that there are no outstanding requests. 967that all threads are killed and that there are no outstanding requests.
861 968
862Under normal circumstances you don't need to call this function. 969Under normal circumstances you don't need to call this function.
863 970
864=item $oldnreqs = IO::AIO::max_outstanding $nreqs 971=item $oldmaxreqs = IO::AIO::max_outstanding $maxreqs
865 972
866[DEPRECATED] 973This is a very bad function to use in interactive programs because it
974blocks, and a bad way to reduce concurrency because it is inexact: Better
975use an C<aio_group> together with a feed callback.
867 976
868Sets the maximum number of outstanding requests to C<$nreqs>. If you 977Sets 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 978to queue up more than this number of requests, the next call to the
870some requests have been handled. 979C<poll_cb> (and C<poll_some> and other functions calling C<poll_cb>)
980function will block until the limit is no longer exceeded.
871 981
872The default is very large, so normally there is no practical limit. If you 982The 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 983number of outstanding requests.
874this to a relatively low number, such as C<100>.
875 984
876This function does not work well together with C<aio_group>'s, and their 985You can still queue as many requests as you want. Therefore,
877feeder interface is better suited to limiting concurrency, so do not use 986C<max_oustsanding> is mainly useful in simple scripts (with low values) or
878this function. 987as 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 988
882=back 989=back
883 990
884=cut 991=cut
885 992
900} 1007}
901 1008
902min_parallel 8; 1009min_parallel 8;
903 1010
904END { 1011END {
905 max_parallel 0; 1012 min_parallel 1;
906} 1013 flush;
1014};
907 1015
9081; 10161;
909 1017
910=head2 FORK BEHAVIOUR 1018=head2 FORK BEHAVIOUR
911 1019

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines