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.41 by root, Wed Sep 7 17:41:17 2005 UTC vs.
Revision 1.54 by root, Sun Oct 22 00:19:05 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
24
25 # AnyEvent
26 open my $fh, "<&=" . IO::AIO::poll_fileno or die "$!";
27 my $w = AnyEvent->io (fh => $fh, poll => 'r', cb => sub { IO::AIO::poll_cb });
19 28
20 # Event 29 # Event
21 Event->io (fd => IO::AIO::poll_fileno, 30 Event->io (fd => IO::AIO::poll_fileno,
22 poll => 'r', 31 poll => 'r',
23 cb => \&IO::AIO::poll_cb); 32 cb => \&IO::AIO::poll_cb);
57=cut 66=cut
58 67
59package IO::AIO; 68package IO::AIO;
60 69
61no warnings; 70no warnings;
71use strict 'vars';
62 72
63use base 'Exporter'; 73use base 'Exporter';
64 74
65use Fcntl ();
66
67BEGIN { 75BEGIN {
68 $VERSION = '1.61'; 76 our $VERSION = '1.99';
69 77
70 @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
71 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
72 aio_fsync aio_fdatasync aio_readahead); 80 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
81 aio_group);
73 @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);
74 max_outstanding nreqs); 83
84 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
75 85
76 require XSLoader; 86 require XSLoader;
77 XSLoader::load IO::AIO, $VERSION; 87 XSLoader::load ("IO::AIO", $VERSION);
78} 88}
79 89
80=head1 FUNCTIONS 90=head1 FUNCTIONS
81 91
82=head2 AIO FUNCTIONS 92=head2 AIO FUNCTIONS
89perl, 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
90syscall has been executed asynchronously. 100syscall has been executed asynchronously.
91 101
92All functions expecting a filehandle keep a copy of the filehandle 102All functions expecting a filehandle keep a copy of the filehandle
93internally until the request has finished. 103internally until the request has finished.
104
105All non-composite requests (requests that are not broken down into
106multiple requests) return objects of type L<IO::AIO::REQ> that allow
107further manipulation of running requests.
94 108
95The pathnames you pass to these routines I<must> be absolute and 109The pathnames you pass to these routines I<must> be absolute and
96encoded in byte form. The reason for the former is that at the time the 110encoded in byte form. The reason for the former is that at the time the
97request is being executed, the current working directory could have 111request is being executed, the current working directory could have
98changed. Alternatively, you can make sure that you never change the 112changed. Alternatively, you can make sure that you never change the
164 aio_read $fh, 7, 15, $buffer, 0, sub { 178 aio_read $fh, 7, 15, $buffer, 0, sub {
165 $_[0] > 0 or die "read error: $!"; 179 $_[0] > 0 or die "read error: $!";
166 print "read $_[0] bytes: <$buffer>\n"; 180 print "read $_[0] bytes: <$buffer>\n";
167 }; 181 };
168 182
183=item aio_move $srcpath, $dstpath, $callback->($status)
184
185Try to move the I<file> (directories not supported as either source or
186destination) from C<$srcpath> to C<$dstpath> and call the callback with
187the C<0> (error) or C<-1> ok.
188
189This is a composite request that tries to rename(2) the file first. If
190rename files with C<EXDEV>, it creates the destination file with mode 0200
191and copies the contents of the source file into it using C<aio_sendfile>,
192followed by restoring atime, mtime, access mode and uid/gid, in that
193order, and unlinking the C<$srcpath>.
194
195If an error occurs, the partial destination file will be unlinked, if
196possible, except when setting atime, mtime, access mode and uid/gid, where
197errors are being ignored.
198
199=cut
200
201sub aio_move($$$) {
202 my ($src, $dst, $cb) = @_;
203
204 aio_rename $src, $dst, sub {
205 if ($_[0] && $! == EXDEV) {
206 aio_open $src, O_RDONLY, 0, sub {
207 if (my $src_fh = $_[0]) {
208 my @stat = stat $src_fh;
209
210 aio_open $dst, O_WRONLY, 0200, sub {
211 if (my $dst_fh = $_[0]) {
212 aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
213 close $src_fh;
214
215 if ($_[0] == $stat[7]) {
216 utime $stat[8], $stat[9], $dst;
217 chmod $stat[2] & 07777, $dst_fh;
218 chown $stat[4], $stat[5], $dst_fh;
219 close $dst_fh;
220
221 aio_unlink $src, sub {
222 $cb->($_[0]);
223 };
224 } else {
225 my $errno = $!;
226 aio_unlink $dst, sub {
227 $! = $errno;
228 $cb->(-1);
229 };
230 }
231 };
232 } else {
233 $cb->(-1);
234 }
235 },
236
237 } else {
238 $cb->(-1);
239 }
240 };
241 } else {
242 $cb->($_[0]);
243 }
244 };
245}
246
169=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval) 247=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
170 248
171Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts 249Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
172reading at byte offset C<$in_offset>, and starts writing at the current 250reading at byte offset C<$in_offset>, and starts writing at the current
173file offset of C<$out_fh>. Because of that, it is not safe to issue more 251file offset of C<$out_fh>. Because of that, it is not safe to issue more
228=item aio_unlink $pathname, $callback->($status) 306=item aio_unlink $pathname, $callback->($status)
229 307
230Asynchronously unlink (delete) a file and call the callback with the 308Asynchronously unlink (delete) a file and call the callback with the
231result code. 309result code.
232 310
311=item aio_link $srcpath, $dstpath, $callback->($status)
312
313Asynchronously create a new link to the existing object at C<$srcpath> at
314the path C<$dstpath> and call the callback with the result code.
315
316=item aio_symlink $srcpath, $dstpath, $callback->($status)
317
318Asynchronously create a new symbolic link to the existing object at C<$srcpath> at
319the path C<$dstpath> and call the callback with the result code.
320
321=item aio_rename $srcpath, $dstpath, $callback->($status)
322
323Asynchronously rename the object at C<$srcpath> to C<$dstpath>, just as
324rename(2) and call the callback with the result code.
325
233=item aio_rmdir $pathname, $callback->($status) 326=item aio_rmdir $pathname, $callback->($status)
234 327
235Asynchronously rmdir (delete) a directory and call the callback with the 328Asynchronously rmdir (delete) a directory and call the callback with the
236result code. 329result code.
237 330
238=item aio_readdir $pathname $callback->($entries) 331=item aio_readdir $pathname, $callback->($entries)
239 332
240Unlike the POSIX call of the same name, C<aio_readdir> reads an entire 333Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
241directory (i.e. opendir + readdir + closedir). The entries will not be 334directory (i.e. opendir + readdir + closedir). The entries will not be
242sorted, and will B<NOT> include the C<.> and C<..> entries. 335sorted, and will B<NOT> include the C<.> and C<..> entries.
243 336
244The callback a single argument which is either C<undef> or an array-ref 337The callback a single argument which is either C<undef> or an array-ref
245with the filenames. 338with the filenames.
246 339
247=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 340=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
248 341
249Scans a directory (similar to C<aio_readdir>) and tries to separate the 342Scans a directory (similar to C<aio_readdir>) but additionally tries to
250entries of directory C<$path> into two sets of names, ones you can recurse 343separate the entries of directory C<$path> into two sets of names, ones
251into (directories), and ones you cannot recurse into (everything else). 344you can recurse into (directories or links to them), and ones you cannot
345recurse into (everything else).
252 346
253C<aio_scandir> is a composite request that consists of many 347C<aio_scandir> is a composite request that consists of many sub
254aio-primitives. C<$maxreq> specifies the maximum number of outstanding 348requests. C<$maxreq> specifies the maximum number of outstanding aio
255aio requests that this function generates. If it is C<< <= 0 >>, then a 349requests that this function generates. If it is C<< <= 0 >>, then a
256suitable default will be chosen (currently 8). 350suitable default will be chosen (currently 8).
257 351
258On error, the callback is called without arguments, otherwise it receives 352On error, the callback is called without arguments, otherwise it receives
259two array-refs with path-relative entry names. 353two array-refs with path-relative entry names.
260 354
269Implementation notes. 363Implementation notes.
270 364
271The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can. 365The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.
272 366
273After reading the directory, the modification time, size etc. of the 367After reading the directory, the modification time, size etc. of the
274directory before and after the readdir is checked, and if they match, the 368directory before and after the readdir is checked, and if they match (and
275link count will be used to decide how many entries are directories (if 369isn't the current time), the link count will be used to decide how many
276>= 2). Otherwise, no knowledge of the number of subdirectories will be 370entries are directories (if >= 2). Otherwise, no knowledge of the number
277assumed. 371of subdirectories will be assumed.
278 372
279Then entires will be sorted into likely directories (everything without a 373Then entries will be sorted into likely directories (everything without
280non-initial dot) and likely non-directories (everything else). Then every 374a non-initial dot currently) and likely non-directories (everything
281entry + C</.> will be C<stat>'ed, likely directories first. This is often 375else). Then every entry plus an appended C</.> will be C<stat>'ed,
376likely directories first. If that succeeds, it assumes that the entry
377is a directory or a symlink to directory (which will be checked
378seperately). This is often faster than stat'ing the entry itself because
282faster because filesystems might detect the type of the entry without 379filesystems might detect the type of the entry without reading the inode
283reading the inode data (e.g. ext2s filetype feature). If that succeeds, 380data (e.g. ext2fs filetype feature).
284it assumes that the entry is a directory or a symlink to directory (which
285will be checked seperately).
286 381
287If the known number of directories has been reached, the rest of the 382If the known number of directories (link count - 2) has been reached, the
288entries is assumed to be non-directories. 383rest of the entries is assumed to be non-directories.
384
385This only works with certainty on POSIX (= UNIX) filesystems, which
386fortunately are the vast majority of filesystems around.
387
388It will also likely work on non-POSIX filesystems with reduced efficiency
389as those tend to return 0 or 1 as link counts, which disables the
390directory counting heuristic.
289 391
290=cut 392=cut
291 393
292sub aio_scandir($$$) { 394sub aio_scandir($$$) {
293 my ($path, $maxreq, $cb) = @_; 395 my ($path, $maxreq, $cb) = @_;
294 396
295 $maxreq = 8 if $maxreq <= 0; 397 $maxreq = 8 if $maxreq <= 0;
296 398
297 # stat once 399 # stat once
298 aio_stat $path, sub { 400 aio_stat $path, sub {
299 $cb->() if $_[0]; 401 return $cb->() if $_[0];
402 my $now = time;
300 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 403 my $hash1 = join ":", (stat _)[0,1,3,7,9];
301 404
302 # read the directory entries 405 # read the directory entries
303 aio_readdir $path, sub { 406 aio_readdir $path, sub {
304 my $entries = shift 407 my $entries = shift
309 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 412 my $hash2 = join ":", (stat _)[0,1,3,7,9];
310 413
311 my $ndirs; 414 my $ndirs;
312 415
313 # take the slow route if anything looks fishy 416 # take the slow route if anything looks fishy
314 if ($hash1 ne $hash2) { 417 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
315 $ndirs = -1; 418 $ndirs = -1;
316 } else { 419 } else {
317 # if nlink == 2, we are finished 420 # if nlink == 2, we are finished
318 # on non-posix-fs's, we rely on nlink < 2 421 # on non-posix-fs's, we rely on nlink < 2
319 $ndirs = (stat _)[3] - 2 422 $ndirs = (stat _)[3] - 2
320 or $cb->([], $entries); 423 or return $cb->([], $entries);
321 } 424 }
322 425
323 # sort into likely dirs and likely nondirs 426 # sort into likely dirs and likely nondirs
324 # dirs == files without ".", short entries first 427 # dirs == files without ".", short entries first
325 $entries = [map $_->[0], 428 $entries = [map $_->[0],
341 } 444 }
342 } elsif (!$nreq) { 445 } elsif (!$nreq) {
343 # finished 446 # finished
344 undef $statcb; 447 undef $statcb;
345 undef $schedcb; 448 undef $schedcb;
346 $cb->(\@dirs, \@nondirs); 449 $cb->(\@dirs, \@nondirs) if $cb;
347 undef $cb; 450 undef $cb;
348 } 451 }
349 }; 452 };
350 $statcb = sub { 453 $statcb = sub {
351 my ($status, $entry) = @_; 454 my ($status, $entry) = @_;
392callback with the fdatasync result code. 495callback with the fdatasync result code.
393 496
394If this call isn't available because your OS lacks it or it couldn't be 497If this call isn't available because your OS lacks it or it couldn't be
395detected, it will be emulated by calling C<fsync> instead. 498detected, it will be emulated by calling C<fsync> instead.
396 499
500=item aio_group $callback->()
501
502=item aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
503
504Mainly used for debugging and benchmarking, this aio request puts one of
505the request workers to sleep for the given time.
506
507=back
508
509=head2 IO::AIO::REQ CLASS
510
511All non-aggregate C<aio_*> functions return an object of this class when
512called in non-void context.
513
514A request always moves through the following five states in its lifetime,
515in order: B<ready> (request has been created, but has not been executed
516yet), B<execute> (request is currently being executed), B<pending>
517(request has been executed but callback has not been called yet),
518B<result> (results are being processed synchronously, includes calling the
519callback) and B<done> (request has reached the end of its lifetime and
520holds no resources anymore).
521
522=over 4
523
524=item $req->cancel
525
526Cancels the request, if possible. Has the effect of skipping execution
527when entering the B<execute> state and skipping calling the callback when
528entering the the B<result> state, but will leave the request otherwise
529untouched. That means that requests that currently execute will not be
530stopped and resources held by the request will not be freed prematurely.
531
397=back 532=back
398 533
399=head2 SUPPORT FUNCTIONS 534=head2 SUPPORT FUNCTIONS
400 535
401=over 4 536=over 4
528} 663}
529 664
5301; 6651;
531 666
532=head2 FORK BEHAVIOUR 667=head2 FORK BEHAVIOUR
668
669This module should do "the right thing" when the process using it forks:
533 670
534Before the fork, IO::AIO enters a quiescent state where no requests 671Before the fork, IO::AIO enters a quiescent state where no requests
535can be added in other threads and no results will be processed. After 672can be added in other threads and no results will be processed. After
536the fork the parent simply leaves the quiescent state and continues 673the fork the parent simply leaves the quiescent state and continues
537request/result processing, while the child clears the request/result 674request/result processing, while the child clears the request/result
538queue (so the requests started before the fork will only be handled in 675queue (so the requests started before the fork will only be handled in
539the parent). Threats will be started on demand until the limit ste in the 676the parent). Threads will be started on demand until the limit ste in the
540parent process has been reached again. 677parent process has been reached again.
541 678
679In short: the parent will, after a short pause, continue as if fork had
680not been called, while the child will act as if IO::AIO has not been used
681yet.
682
542=head1 SEE ALSO 683=head1 SEE ALSO
543 684
544L<Coro>, L<Linux::AIO>. 685L<Coro>, L<Linux::AIO> (obsolete).
545 686
546=head1 AUTHOR 687=head1 AUTHOR
547 688
548 Marc Lehmann <schmorp@schmorp.de> 689 Marc Lehmann <schmorp@schmorp.de>
549 http://home.schmorp.de/ 690 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines