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.50 by root, Sat Jun 24 16:27:02 2006 UTC vs.
Revision 1.55 by root, Sun Oct 22 00:49:29 2006 UTC

14 aio_unlink "/tmp/file", sub { }; 14 aio_unlink "/tmp/file", sub { };
15 15
16 aio_read $fh, 30000, 1024, $buffer, 0, sub { 16 aio_read $fh, 30000, 1024, $buffer, 0, sub {
17 $_[0] > 0 or die "read error: $!"; 17 $_[0] > 0 or die "read error: $!";
18 }; 18 };
19
20 use IO::AIO 2; # version has aio objects
21
22 my $req = aio_unlink "/tmp/file", sub { };
23 $req->cancel; # cancel request if still in queue
19 24
20 # AnyEvent 25 # AnyEvent
21 open my $fh, "<&=" . IO::AIO::poll_fileno or die "$!"; 26 open my $fh, "<&=" . IO::AIO::poll_fileno or die "$!";
22 my $w = AnyEvent->io (fh => $fh, poll => 'r', cb => sub { IO::AIO::poll_cb }); 27 my $w = AnyEvent->io (fh => $fh, poll => 'r', cb => sub { IO::AIO::poll_cb });
23 28
61=cut 66=cut
62 67
63package IO::AIO; 68package IO::AIO;
64 69
65no warnings; 70no warnings;
71use strict 'vars';
66 72
67use base 'Exporter'; 73use base 'Exporter';
68 74
69use Fcntl ();
70
71BEGIN { 75BEGIN {
72 $VERSION = '1.8'; 76 our $VERSION = '2.0';
73 77
74 @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 78 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
75 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink 79 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
76 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move); 80 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
81 aio_group);
77 @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel 82 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs);
78 max_outstanding nreqs); 83
84 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
79 85
80 require XSLoader; 86 require XSLoader;
81 XSLoader::load IO::AIO, $VERSION; 87 XSLoader::load ("IO::AIO", $VERSION);
82} 88}
83 89
84=head1 FUNCTIONS 90=head1 FUNCTIONS
85 91
86=head2 AIO FUNCTIONS 92=head2 AIO FUNCTIONS
93perl, which usually delivers "false") as it's sole argument when the given 99perl, which usually delivers "false") as it's sole argument when the given
94syscall has been executed asynchronously. 100syscall has been executed asynchronously.
95 101
96All functions expecting a filehandle keep a copy of the filehandle 102All functions expecting a filehandle keep a copy of the filehandle
97internally until the request has finished. 103internally until the request has finished.
104
105All requests return objects of type L<IO::AIO::REQ> that allow further
106manipulation of those requests while they are in-flight.
98 107
99The pathnames you pass to these routines I<must> be absolute and 108The pathnames you pass to these routines I<must> be absolute and
100encoded in byte form. The reason for the former is that at the time the 109encoded in byte form. The reason for the former is that at the time the
101request is being executed, the current working directory could have 110request is being executed, the current working directory could have
102changed. Alternatively, you can make sure that you never change the 111changed. Alternatively, you can make sure that you never change the
170 print "read $_[0] bytes: <$buffer>\n"; 179 print "read $_[0] bytes: <$buffer>\n";
171 }; 180 };
172 181
173=item aio_move $srcpath, $dstpath, $callback->($status) 182=item aio_move $srcpath, $dstpath, $callback->($status)
174 183
175[EXPERIMENTAL]
176
177Try to move the I<file> (directories not supported as either source or destination) 184Try to move the I<file> (directories not supported as either source or
178from C<$srcpath> to C<$dstpath> and call the callback with the C<0> (error) or C<-1> ok. 185destination) from C<$srcpath> to C<$dstpath> and call the callback with
186the C<0> (error) or C<-1> ok.
179 187
180This is a composite request that tries to rename(2) the file first. If 188This is a composite request that tries to rename(2) the file first. If
181rename files with C<EXDEV>, it creates the destination file with mode 0200 189rename files with C<EXDEV>, it creates the destination file with mode 0200
182and copies the contents of the source file into it using C<aio_sendfile>, 190and copies the contents of the source file into it using C<aio_sendfile>,
183followed by restoring atime, mtime, access mode and uid/gid, in that 191followed by restoring atime, mtime, access mode and uid/gid, in that
190=cut 198=cut
191 199
192sub aio_move($$$) { 200sub aio_move($$$) {
193 my ($src, $dst, $cb) = @_; 201 my ($src, $dst, $cb) = @_;
194 202
203 my $grp = aio_group;
204
195 aio_rename $src, $dst, sub { 205 add $grp aio_rename $src, $dst, sub {
196 if ($_[0] && $! == Errno::EXDEV) { 206 if ($_[0] && $! == EXDEV) {
197 aio_open $src, O_RDONLY, 0, sub { 207 add $grp aio_open $src, O_RDONLY, 0, sub {
198 if (my $src_fh = $_[0]) { 208 if (my $src_fh = $_[0]) {
199 my @stat = stat $src_fh; 209 my @stat = stat $src_fh;
200 210
201 aio_open $dst, O_WRONLY, 0200, sub { 211 add $grp aio_open $dst, O_WRONLY, 0200, sub {
202 if (my $dst_fh = $_[0]) { 212 if (my $dst_fh = $_[0]) {
203 aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub { 213 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
204 close $src_fh; 214 close $src_fh;
205 215
206 if ($_[0] == $stat[7]) { 216 if ($_[0] == $stat[7]) {
207 utime $stat[8], $stat[9], $dst; 217 utime $stat[8], $stat[9], $dst;
208 chmod $stat[2] & 07777, $dst_fh; 218 chmod $stat[2] & 07777, $dst_fh;
209 chown $stat[4], $stat[5], $dst_fh; 219 chown $stat[4], $stat[5], $dst_fh;
210 close $dst_fh; 220 close $dst_fh;
211 221
212 aio_unlink $src, sub { 222 add $grp aio_unlink $src, sub {
213 $cb->($_[0]); 223 $cb->($_[0]);
214 }; 224 };
215 } else { 225 } else {
216 my $errno = $!; 226 my $errno = $!;
217 aio_unlink $dst, sub { 227 add $grp aio_unlink $dst, sub {
218 $! = $errno; 228 $! = $errno;
219 $cb->(-1); 229 $cb->(-1);
220 }; 230 };
221 } 231 }
222 }; 232 };
231 }; 241 };
232 } else { 242 } else {
233 $cb->($_[0]); 243 $cb->($_[0]);
234 } 244 }
235 }; 245 };
246
247 $grp
236} 248}
237 249
238=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval) 250=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
239 251
240Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts 252Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
328The callback a single argument which is either C<undef> or an array-ref 340The callback a single argument which is either C<undef> or an array-ref
329with the filenames. 341with the filenames.
330 342
331=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 343=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
332 344
333Scans a directory (similar to C<aio_readdir>) and tries to separate the 345Scans a directory (similar to C<aio_readdir>) but additionally tries to
334entries of directory C<$path> into two sets of names, ones you can recurse 346separate the entries of directory C<$path> into two sets of names, ones
335into (directories), and ones you cannot recurse into (everything else). 347you can recurse into (directories or links to them), and ones you cannot
348recurse into (everything else).
336 349
337C<aio_scandir> is a composite request that consists of many 350C<aio_scandir> is a composite request that consists of many sub
338aio-primitives. C<$maxreq> specifies the maximum number of outstanding 351requests. C<$maxreq> specifies the maximum number of outstanding aio
339aio requests that this function generates. If it is C<< <= 0 >>, then a 352requests that this function generates. If it is C<< <= 0 >>, then a
340suitable default will be chosen (currently 8). 353suitable default will be chosen (currently 8).
341 354
342On error, the callback is called without arguments, otherwise it receives 355On error, the callback is called without arguments, otherwise it receives
343two array-refs with path-relative entry names. 356two array-refs with path-relative entry names.
344 357
353Implementation notes. 366Implementation notes.
354 367
355The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can. 368The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.
356 369
357After reading the directory, the modification time, size etc. of the 370After reading the directory, the modification time, size etc. of the
358directory before and after the readdir is checked, and if they match, the 371directory before and after the readdir is checked, and if they match (and
359link count will be used to decide how many entries are directories (if 372isn't the current time), the link count will be used to decide how many
360>= 2). Otherwise, no knowledge of the number of subdirectories will be 373entries are directories (if >= 2). Otherwise, no knowledge of the number
361assumed. 374of subdirectories will be assumed.
362 375
363Then entires will be sorted into likely directories (everything without a 376Then entries will be sorted into likely directories (everything without
364non-initial dot) and likely non-directories (everything else). Then every 377a non-initial dot currently) and likely non-directories (everything
365entry + C</.> will be C<stat>'ed, likely directories first. This is often 378else). Then every entry plus an appended C</.> will be C<stat>'ed,
379likely directories first. If that succeeds, it assumes that the entry
380is a directory or a symlink to directory (which will be checked
381seperately). This is often faster than stat'ing the entry itself because
366faster because filesystems might detect the type of the entry without 382filesystems might detect the type of the entry without reading the inode
367reading the inode data (e.g. ext2fs filetype feature). If that succeeds, 383data (e.g. ext2fs filetype feature).
368it assumes that the entry is a directory or a symlink to directory (which
369will be checked seperately).
370 384
371If the known number of directories has been reached, the rest of the 385If the known number of directories (link count - 2) has been reached, the
372entries is assumed to be non-directories. 386rest of the entries is assumed to be non-directories.
387
388This only works with certainty on POSIX (= UNIX) filesystems, which
389fortunately are the vast majority of filesystems around.
390
391It will also likely work on non-POSIX filesystems with reduced efficiency
392as those tend to return 0 or 1 as link counts, which disables the
393directory counting heuristic.
373 394
374=cut 395=cut
375 396
376sub aio_scandir($$$) { 397sub aio_scandir($$$) {
377 my ($path, $maxreq, $cb) = @_; 398 my ($path, $maxreq, $cb) = @_;
378 399
400 my $grp = aio_group;
401
379 $maxreq = 8 if $maxreq <= 0; 402 $maxreq = 8 if $maxreq <= 0;
380 403
381 # stat once 404 # stat once
382 aio_stat $path, sub { 405 add $grp aio_stat $path, sub {
383 return $cb->() if $_[0]; 406 return $cb->() if $_[0];
407 my $now = time;
384 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 408 my $hash1 = join ":", (stat _)[0,1,3,7,9];
385 409
386 # read the directory entries 410 # read the directory entries
387 aio_readdir $path, sub { 411 add $grp aio_readdir $path, sub {
388 my $entries = shift 412 my $entries = shift
389 or return $cb->(); 413 or return $cb->();
390 414
391 # stat the dir another time 415 # stat the dir another time
392 aio_stat $path, sub { 416 add $grp aio_stat $path, sub {
393 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 417 my $hash2 = join ":", (stat _)[0,1,3,7,9];
394 418
395 my $ndirs; 419 my $ndirs;
396 420
397 # take the slow route if anything looks fishy 421 # take the slow route if anything looks fishy
398 if ($hash1 ne $hash2) { 422 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
399 $ndirs = -1; 423 $ndirs = -1;
400 } else { 424 } else {
401 # if nlink == 2, we are finished 425 # if nlink == 2, we are finished
402 # on non-posix-fs's, we rely on nlink < 2 426 # on non-posix-fs's, we rely on nlink < 2
403 $ndirs = (stat _)[3] - 2 427 $ndirs = (stat _)[3] - 2
419 $schedcb = sub { 443 $schedcb = sub {
420 if (@$entries) { 444 if (@$entries) {
421 if ($nreq < $maxreq) { 445 if ($nreq < $maxreq) {
422 my $ent = pop @$entries; 446 my $ent = pop @$entries;
423 $nreq++; 447 $nreq++;
424 aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) }; 448 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
425 } 449 }
426 } elsif (!$nreq) { 450 } elsif (!$nreq) {
427 # finished 451 # finished
428 undef $statcb; 452 undef $statcb;
429 undef $schedcb; 453 undef $schedcb;
438 $nreq--; 462 $nreq--;
439 push @nondirs, $entry; 463 push @nondirs, $entry;
440 &$schedcb; 464 &$schedcb;
441 } else { 465 } else {
442 # need to check for real directory 466 # need to check for real directory
443 aio_lstat "$path/$entry", sub { 467 add $grp aio_lstat "$path/$entry", sub {
444 $nreq--; 468 $nreq--;
445 469
446 if (-d _) { 470 if (-d _) {
447 push @dirs, $entry; 471 push @dirs, $entry;
448 472
461 485
462 &$schedcb while @$entries && $nreq < $maxreq; 486 &$schedcb while @$entries && $nreq < $maxreq;
463 }; 487 };
464 }; 488 };
465 }; 489 };
490
491 $grp
466} 492}
467 493
468=item aio_fsync $fh, $callback->($status) 494=item aio_fsync $fh, $callback->($status)
469 495
470Asynchronously call fsync on the given filehandle and call the callback 496Asynchronously call fsync on the given filehandle and call the callback
475Asynchronously call fdatasync on the given filehandle and call the 501Asynchronously call fdatasync on the given filehandle and call the
476callback with the fdatasync result code. 502callback with the fdatasync result code.
477 503
478If this call isn't available because your OS lacks it or it couldn't be 504If this call isn't available because your OS lacks it or it couldn't be
479detected, it will be emulated by calling C<fsync> instead. 505detected, it will be emulated by calling C<fsync> instead.
506
507=item aio_group $callback->()
508
509[EXPERIMENTAL]
510
511This is a very special aio request: Instead of doing something, it is a
512container for other aio requests, which is useful if you want to bundle
513many requests into a single, composite, request.
514
515Returns an object of class L<IO::AIO::GRP>. See its documentation below
516for more info.
517
518Example:
519
520 my $grp = aio_group sub {
521 print "all stats done\n";
522 };
523
524 add $grp
525 (aio_stat ...),
526 (aio_stat ...),
527 ...;
528
529=item aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
530
531Mainly used for debugging and benchmarking, this aio request puts one of
532the request workers to sleep for the given time.
533
534=back
535
536=head2 IO::AIO::REQ CLASS
537
538All non-aggregate C<aio_*> functions return an object of this class when
539called in non-void context.
540
541A request always moves through the following five states in its lifetime,
542in order: B<ready> (request has been created, but has not been executed
543yet), B<execute> (request is currently being executed), B<pending>
544(request has been executed but callback has not been called yet),
545B<result> (results are being processed synchronously, includes calling the
546callback) and B<done> (request has reached the end of its lifetime and
547holds no resources anymore).
548
549=over 4
550
551=item $req->cancel
552
553Cancels the request, if possible. Has the effect of skipping execution
554when entering the B<execute> state and skipping calling the callback when
555entering the the B<result> state, but will leave the request otherwise
556untouched. That means that requests that currently execute will not be
557stopped and resources held by the request will not be freed prematurely.
558
559=back
560
561=head2 IO::AIO::GRP CLASS
562
563This class is a subclass of L<IO::AIO::REQ>, so all its methods apply to
564objects of this class, too.
565
566A IO::AIO::GRP object is a special request that can contain multiple other
567aio requests.
568
569You create one by calling the C<aio_group> constructing function with a
570callback that will be called when all contained requests have entered the
571C<done> state:
572
573 my $grp = aio_group sub {
574 print "all requests are done\n";
575 };
576
577You add requests by calling the C<add> method with one or more
578C<IO::AIO::REQ> objects:
579
580 $grp->add (aio_unlink "...");
581
582 add $grp aio_stat "...", sub { ... };
583
584This makes it very easy to create composite requests (see the source of
585C<aio_move> for an application) that work and feel like simple requests.
586
587The IO::AIO::GRP objects will be cleaned up during calls to
588C<IO::AIO::poll_cb>, just like any other request.
589
590They can be canceled like any other request. Canceling will cancel not
591just the request itself, but also all requests it contains.
592
593They can also can also be added to other IO::AIO::GRP objects.
594
595Their lifetime, simplified, looks like this: when they are empty, they
596will finish very quickly. If they contain only requests that are in the
597C<done> state, they will also finish. Otherwise they will continue to
598exist.
599
600=over 4
601
602=item $grp->add (...)
603
604=item add $grp ...
605
606Add one or more
607Cancels the request, if possible. Has the effect of skipping execution
608when entering the B<execute> state and skipping calling the callback when
609entering the the B<result> state, but will leave the request otherwise
610untouched. That means that requests that currently execute will not be
611stopped and resources held by the request will not be freed prematurely.
480 612
481=back 613=back
482 614
483=head2 SUPPORT FUNCTIONS 615=head2 SUPPORT FUNCTIONS
484 616
612} 744}
613 745
6141; 7461;
615 747
616=head2 FORK BEHAVIOUR 748=head2 FORK BEHAVIOUR
749
750This module should do "the right thing" when the process using it forks:
617 751
618Before the fork, IO::AIO enters a quiescent state where no requests 752Before the fork, IO::AIO enters a quiescent state where no requests
619can be added in other threads and no results will be processed. After 753can be added in other threads and no results will be processed. After
620the fork the parent simply leaves the quiescent state and continues 754the fork the parent simply leaves the quiescent state and continues
621request/result processing, while the child clears the request/result 755request/result processing, while the child clears the request/result
622queue (so the requests started before the fork will only be handled in 756queue (so the requests started before the fork will only be handled in
623the parent). Threats will be started on demand until the limit ste in the 757the parent). Threads will be started on demand until the limit ste in the
624parent process has been reached again. 758parent process has been reached again.
625 759
760In short: the parent will, after a short pause, continue as if fork had
761not been called, while the child will act as if IO::AIO has not been used
762yet.
763
626=head1 SEE ALSO 764=head1 SEE ALSO
627 765
628L<Coro>, L<Linux::AIO>. 766L<Coro>, L<Linux::AIO> (obsolete).
629 767
630=head1 AUTHOR 768=head1 AUTHOR
631 769
632 Marc Lehmann <schmorp@schmorp.de> 770 Marc Lehmann <schmorp@schmorp.de>
633 http://home.schmorp.de/ 771 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines