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.48 by root, Thu Feb 16 16:38:57 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.72'; 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); 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
168 aio_read $fh, 7, 15, $buffer, 0, sub { 177 aio_read $fh, 7, 15, $buffer, 0, sub {
169 $_[0] > 0 or die "read error: $!"; 178 $_[0] > 0 or die "read error: $!";
170 print "read $_[0] bytes: <$buffer>\n"; 179 print "read $_[0] bytes: <$buffer>\n";
171 }; 180 };
172 181
182=item aio_move $srcpath, $dstpath, $callback->($status)
183
184Try to move the I<file> (directories not supported as either source or
185destination) from C<$srcpath> to C<$dstpath> and call the callback with
186the C<0> (error) or C<-1> ok.
187
188This is a composite request that tries to rename(2) the file first. If
189rename files with C<EXDEV>, it creates the destination file with mode 0200
190and copies the contents of the source file into it using C<aio_sendfile>,
191followed by restoring atime, mtime, access mode and uid/gid, in that
192order, and unlinking the C<$srcpath>.
193
194If an error occurs, the partial destination file will be unlinked, if
195possible, except when setting atime, mtime, access mode and uid/gid, where
196errors are being ignored.
197
198=cut
199
200sub aio_move($$$) {
201 my ($src, $dst, $cb) = @_;
202
203 my $grp = aio_group;
204
205 add $grp aio_rename $src, $dst, sub {
206 if ($_[0] && $! == EXDEV) {
207 add $grp aio_open $src, O_RDONLY, 0, sub {
208 if (my $src_fh = $_[0]) {
209 my @stat = stat $src_fh;
210
211 add $grp aio_open $dst, O_WRONLY, 0200, sub {
212 if (my $dst_fh = $_[0]) {
213 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
214 close $src_fh;
215
216 if ($_[0] == $stat[7]) {
217 utime $stat[8], $stat[9], $dst;
218 chmod $stat[2] & 07777, $dst_fh;
219 chown $stat[4], $stat[5], $dst_fh;
220 close $dst_fh;
221
222 add $grp aio_unlink $src, sub {
223 $cb->($_[0]);
224 };
225 } else {
226 my $errno = $!;
227 add $grp aio_unlink $dst, sub {
228 $! = $errno;
229 $cb->(-1);
230 };
231 }
232 };
233 } else {
234 $cb->(-1);
235 }
236 },
237
238 } else {
239 $cb->(-1);
240 }
241 };
242 } else {
243 $cb->($_[0]);
244 }
245 };
246
247 $grp
248}
249
173=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)
174 251
175Tries 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
176reading at byte offset C<$in_offset>, and starts writing at the current 253reading at byte offset C<$in_offset>, and starts writing at the current
177file offset of C<$out_fh>. Because of that, it is not safe to issue more 254file offset of C<$out_fh>. Because of that, it is not safe to issue more
232=item aio_unlink $pathname, $callback->($status) 309=item aio_unlink $pathname, $callback->($status)
233 310
234Asynchronously unlink (delete) a file and call the callback with the 311Asynchronously unlink (delete) a file and call the callback with the
235result code. 312result code.
236 313
314=item aio_link $srcpath, $dstpath, $callback->($status)
315
316Asynchronously create a new link to the existing object at C<$srcpath> at
317the path C<$dstpath> and call the callback with the result code.
318
319=item aio_symlink $srcpath, $dstpath, $callback->($status)
320
321Asynchronously create a new symbolic link to the existing object at C<$srcpath> at
322the path C<$dstpath> and call the callback with the result code.
323
324=item aio_rename $srcpath, $dstpath, $callback->($status)
325
326Asynchronously rename the object at C<$srcpath> to C<$dstpath>, just as
327rename(2) and call the callback with the result code.
328
237=item aio_rmdir $pathname, $callback->($status) 329=item aio_rmdir $pathname, $callback->($status)
238 330
239Asynchronously rmdir (delete) a directory and call the callback with the 331Asynchronously rmdir (delete) a directory and call the callback with the
240result code. 332result code.
241 333
248The 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
249with the filenames. 341with the filenames.
250 342
251=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 343=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
252 344
253Scans a directory (similar to C<aio_readdir>) and tries to separate the 345Scans a directory (similar to C<aio_readdir>) but additionally tries to
254entries 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
255into (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).
256 349
257C<aio_scandir> is a composite request that consists of many 350C<aio_scandir> is a composite request that consists of many sub
258aio-primitives. C<$maxreq> specifies the maximum number of outstanding 351requests. C<$maxreq> specifies the maximum number of outstanding aio
259aio requests that this function generates. If it is C<< <= 0 >>, then a 352requests that this function generates. If it is C<< <= 0 >>, then a
260suitable default will be chosen (currently 8). 353suitable default will be chosen (currently 8).
261 354
262On error, the callback is called without arguments, otherwise it receives 355On error, the callback is called without arguments, otherwise it receives
263two array-refs with path-relative entry names. 356two array-refs with path-relative entry names.
264 357
273Implementation notes. 366Implementation notes.
274 367
275The 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.
276 369
277After reading the directory, the modification time, size etc. of the 370After reading the directory, the modification time, size etc. of the
278directory 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
279link 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
280>= 2). Otherwise, no knowledge of the number of subdirectories will be 373entries are directories (if >= 2). Otherwise, no knowledge of the number
281assumed. 374of subdirectories will be assumed.
282 375
283Then entires will be sorted into likely directories (everything without a 376Then entries will be sorted into likely directories (everything without
284non-initial dot) and likely non-directories (everything else). Then every 377a non-initial dot currently) and likely non-directories (everything
285entry + 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
286faster because filesystems might detect the type of the entry without 382filesystems might detect the type of the entry without reading the inode
287reading the inode data (e.g. ext2fs filetype feature). If that succeeds, 383data (e.g. ext2fs filetype feature).
288it assumes that the entry is a directory or a symlink to directory (which
289will be checked seperately).
290 384
291If 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
292entries 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.
293 394
294=cut 395=cut
295 396
296sub aio_scandir($$$) { 397sub aio_scandir($$$) {
297 my ($path, $maxreq, $cb) = @_; 398 my ($path, $maxreq, $cb) = @_;
298 399
400 my $grp = aio_group;
401
299 $maxreq = 8 if $maxreq <= 0; 402 $maxreq = 8 if $maxreq <= 0;
300 403
301 # stat once 404 # stat once
302 aio_stat $path, sub { 405 add $grp aio_stat $path, sub {
303 return $cb->() if $_[0]; 406 return $cb->() if $_[0];
407 my $now = time;
304 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 408 my $hash1 = join ":", (stat _)[0,1,3,7,9];
305 409
306 # read the directory entries 410 # read the directory entries
307 aio_readdir $path, sub { 411 add $grp aio_readdir $path, sub {
308 my $entries = shift 412 my $entries = shift
309 or return $cb->(); 413 or return $cb->();
310 414
311 # stat the dir another time 415 # stat the dir another time
312 aio_stat $path, sub { 416 add $grp aio_stat $path, sub {
313 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 417 my $hash2 = join ":", (stat _)[0,1,3,7,9];
314 418
315 my $ndirs; 419 my $ndirs;
316 420
317 # take the slow route if anything looks fishy 421 # take the slow route if anything looks fishy
318 if ($hash1 ne $hash2) { 422 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
319 $ndirs = -1; 423 $ndirs = -1;
320 } else { 424 } else {
321 # if nlink == 2, we are finished 425 # if nlink == 2, we are finished
322 # on non-posix-fs's, we rely on nlink < 2 426 # on non-posix-fs's, we rely on nlink < 2
323 $ndirs = (stat _)[3] - 2 427 $ndirs = (stat _)[3] - 2
339 $schedcb = sub { 443 $schedcb = sub {
340 if (@$entries) { 444 if (@$entries) {
341 if ($nreq < $maxreq) { 445 if ($nreq < $maxreq) {
342 my $ent = pop @$entries; 446 my $ent = pop @$entries;
343 $nreq++; 447 $nreq++;
344 aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) }; 448 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
345 } 449 }
346 } elsif (!$nreq) { 450 } elsif (!$nreq) {
347 # finished 451 # finished
348 undef $statcb; 452 undef $statcb;
349 undef $schedcb; 453 undef $schedcb;
358 $nreq--; 462 $nreq--;
359 push @nondirs, $entry; 463 push @nondirs, $entry;
360 &$schedcb; 464 &$schedcb;
361 } else { 465 } else {
362 # need to check for real directory 466 # need to check for real directory
363 aio_lstat "$path/$entry", sub { 467 add $grp aio_lstat "$path/$entry", sub {
364 $nreq--; 468 $nreq--;
365 469
366 if (-d _) { 470 if (-d _) {
367 push @dirs, $entry; 471 push @dirs, $entry;
368 472
381 485
382 &$schedcb while @$entries && $nreq < $maxreq; 486 &$schedcb while @$entries && $nreq < $maxreq;
383 }; 487 };
384 }; 488 };
385 }; 489 };
490
491 $grp
386} 492}
387 493
388=item aio_fsync $fh, $callback->($status) 494=item aio_fsync $fh, $callback->($status)
389 495
390Asynchronously call fsync on the given filehandle and call the callback 496Asynchronously call fsync on the given filehandle and call the callback
395Asynchronously call fdatasync on the given filehandle and call the 501Asynchronously call fdatasync on the given filehandle and call the
396callback with the fdatasync result code. 502callback with the fdatasync result code.
397 503
398If 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
399detected, 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.
400 612
401=back 613=back
402 614
403=head2 SUPPORT FUNCTIONS 615=head2 SUPPORT FUNCTIONS
404 616
532} 744}
533 745
5341; 7461;
535 747
536=head2 FORK BEHAVIOUR 748=head2 FORK BEHAVIOUR
749
750This module should do "the right thing" when the process using it forks:
537 751
538Before the fork, IO::AIO enters a quiescent state where no requests 752Before the fork, IO::AIO enters a quiescent state where no requests
539can 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
540the fork the parent simply leaves the quiescent state and continues 754the fork the parent simply leaves the quiescent state and continues
541request/result processing, while the child clears the request/result 755request/result processing, while the child clears the request/result
542queue (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
543the 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
544parent process has been reached again. 758parent process has been reached again.
545 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
546=head1 SEE ALSO 764=head1 SEE ALSO
547 765
548L<Coro>, L<Linux::AIO>. 766L<Coro>, L<Linux::AIO> (obsolete).
549 767
550=head1 AUTHOR 768=head1 AUTHOR
551 769
552 Marc Lehmann <schmorp@schmorp.de> 770 Marc Lehmann <schmorp@schmorp.de>
553 http://home.schmorp.de/ 771 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines