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.35 by root, Mon Aug 22 23:20:37 2005 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
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.5; 76 our $VERSION = '2.0';
69 77
70 @EXPORT = qw(aio_read aio_write aio_open aio_close aio_stat aio_lstat aio_unlink 78 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
71 aio_rmdir aio_symlink aio_fsync aio_fdatasync aio_readahead); 79 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
80 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
81 aio_group);
72 @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs); 82 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs);
83
84 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
73 85
74 require XSLoader; 86 require XSLoader;
75 XSLoader::load IO::AIO, $VERSION; 87 XSLoader::load ("IO::AIO", $VERSION);
76} 88}
77 89
78=head1 FUNCTIONS 90=head1 FUNCTIONS
79 91
80=head2 AIO FUNCTIONS 92=head2 AIO FUNCTIONS
88syscall has been executed asynchronously. 100syscall has been executed asynchronously.
89 101
90All functions expecting a filehandle keep a copy of the filehandle 102All functions expecting a filehandle keep a copy of the filehandle
91internally until the request has finished. 103internally until the request has finished.
92 104
105All requests return objects of type L<IO::AIO::REQ> that allow further
106manipulation of those requests while they are in-flight.
107
93The pathnames you pass to these routines I<must> be absolute and 108The pathnames you pass to these routines I<must> be absolute and
94encoded 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
95request is being executed, the current working directory could have 110request is being executed, the current working directory could have
96changed. Alternatively, you can make sure that you never change the 111changed. Alternatively, you can make sure that you never change the
97current working directory. 112current working directory.
103environment, d) use Glib::filename_from_unicode on unicode filenames or e) 118environment, d) use Glib::filename_from_unicode on unicode filenames or e)
104use something else. 119use something else.
105 120
106=over 4 121=over 4
107 122
108=item aio_open $pathname, $flags, $mode, $callback 123=item aio_open $pathname, $flags, $mode, $callback->($fh)
109 124
110Asynchronously open or create a file and call the callback with a newly 125Asynchronously open or create a file and call the callback with a newly
111created filehandle for the file. 126created filehandle for the file.
112 127
113The pathname passed to C<aio_open> must be absolute. See API NOTES, above, 128The pathname passed to C<aio_open> must be absolute. See API NOTES, above,
130 } else { 145 } else {
131 die "open failed: $!\n"; 146 die "open failed: $!\n";
132 } 147 }
133 }; 148 };
134 149
135=item aio_close $fh, $callback 150=item aio_close $fh, $callback->($status)
136 151
137Asynchronously close a file and call the callback with the result 152Asynchronously close a file and call the callback with the result
138code. I<WARNING:> although accepted, you should not pass in a perl 153code. I<WARNING:> although accepted, you should not pass in a perl
139filehandle here, as perl will likely close the file descriptor another 154filehandle here, as perl will likely close the file descriptor another
140time when the filehandle is destroyed. Normally, you can safely call perls 155time when the filehandle is destroyed. Normally, you can safely call perls
141C<close> or just let filehandles go out of scope. 156C<close> or just let filehandles go out of scope.
142 157
143This is supposed to be a bug in the API, so that might change. It's 158This is supposed to be a bug in the API, so that might change. It's
144therefore best to avoid this function. 159therefore best to avoid this function.
145 160
146=item aio_read $fh,$offset,$length, $data,$dataoffset,$callback 161=item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
147 162
148=item aio_write $fh,$offset,$length, $data,$dataoffset,$callback 163=item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
149 164
150Reads or writes C<length> bytes from the specified C<fh> and C<offset> 165Reads or writes C<length> bytes from the specified C<fh> and C<offset>
151into the scalar given by C<data> and offset C<dataoffset> and calls the 166into the scalar given by C<data> and offset C<dataoffset> and calls the
152callback without the actual number of bytes read (or -1 on error, just 167callback without the actual number of bytes read (or -1 on error, just
153like the syscall). 168like the syscall).
162 aio_read $fh, 7, 15, $buffer, 0, sub { 177 aio_read $fh, 7, 15, $buffer, 0, sub {
163 $_[0] > 0 or die "read error: $!"; 178 $_[0] > 0 or die "read error: $!";
164 print "read $_[0] bytes: <$buffer>\n"; 179 print "read $_[0] bytes: <$buffer>\n";
165 }; 180 };
166 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
167=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback 250=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
168 251
169Tries 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
170reading 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
171file 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
172than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each 255than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each
175This call tries to make use of a native C<sendfile> syscall to provide 258This call tries to make use of a native C<sendfile> syscall to provide
176zero-copy operation. For this to work, C<$out_fh> should refer to a 259zero-copy operation. For this to work, C<$out_fh> should refer to a
177socket, and C<$in_fh> should refer to mmap'able file. 260socket, and C<$in_fh> should refer to mmap'able file.
178 261
179If the native sendfile call fails or is not implemented, it will be 262If the native sendfile call fails or is not implemented, it will be
180emulated, so you can call C<aio_sendfile> on any filehandles regardless of 263emulated, so you can call C<aio_sendfile> on any type of filehandle
181the limitations of the OS. 264regardless of the limitations of the operating system.
182 265
183Please note, however, that C<aio_sendfile> can read more bytes from 266Please note, however, that C<aio_sendfile> can read more bytes from
184C<$in_fh> than are written, and there is no way to find out how many 267C<$in_fh> than are written, and there is no way to find out how many
185bytes have been read form C<aio_sendfile> alone, as C<aio_sendfile> only 268bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only
186provides the number of bytes written to C<$out_fh>. Only if the return 269provides the number of bytes written to C<$out_fh>. Only if the result
187value (the value provided to the callback) equals C<$length> one can 270value equals C<$length> one can assume that C<$length> bytes have been
188assume that C<$length> bytes have been read. 271read.
189 272
190=item aio_readahead $fh,$offset,$length, $callback 273=item aio_readahead $fh,$offset,$length, $callback->($retval)
191 274
192C<aio_readahead> populates the page cache with data from a file so that 275C<aio_readahead> populates the page cache with data from a file so that
193subsequent reads from that file will not block on disk I/O. The C<$offset> 276subsequent reads from that file will not block on disk I/O. The C<$offset>
194argument specifies the starting point from which data is to be read and 277argument specifies the starting point from which data is to be read and
195C<$length> specifies the number of bytes to be read. I/O is performed in 278C<$length> specifies the number of bytes to be read. I/O is performed in
199file. The current file offset of the file is left unchanged. 282file. The current file offset of the file is left unchanged.
200 283
201If that syscall doesn't exist (likely if your OS isn't Linux) it will be 284If that syscall doesn't exist (likely if your OS isn't Linux) it will be
202emulated by simply reading the data, which would have a similar effect. 285emulated by simply reading the data, which would have a similar effect.
203 286
204=item aio_stat $fh_or_path, $callback 287=item aio_stat $fh_or_path, $callback->($status)
205 288
206=item aio_lstat $fh, $callback 289=item aio_lstat $fh, $callback->($status)
207 290
208Works like perl's C<stat> or C<lstat> in void context. The callback will 291Works like perl's C<stat> or C<lstat> in void context. The callback will
209be called after the stat and the results will be available using C<stat _> 292be called after the stat and the results will be available using C<stat _>
210or C<-s _> etc... 293or C<-s _> etc...
211 294
221 aio_stat "/etc/passwd", sub { 304 aio_stat "/etc/passwd", sub {
222 $_[0] and die "stat failed: $!"; 305 $_[0] and die "stat failed: $!";
223 print "size is ", -s _, "\n"; 306 print "size is ", -s _, "\n";
224 }; 307 };
225 308
226=item aio_unlink $pathname, $callback 309=item aio_unlink $pathname, $callback->($status)
227 310
228Asynchronously unlink (delete) a file and call the callback with the 311Asynchronously unlink (delete) a file and call the callback with the
229result code. 312result code.
230 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
231=item aio_rmdir $pathname, $callback 329=item aio_rmdir $pathname, $callback->($status)
232 330
233Asynchronously rmdir (delete) a directory and call the callback with the 331Asynchronously rmdir (delete) a directory and call the callback with the
234result code. 332result code.
235 333
334=item aio_readdir $pathname, $callback->($entries)
335
336Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
337directory (i.e. opendir + readdir + closedir). The entries will not be
338sorted, and will B<NOT> include the C<.> and C<..> entries.
339
340The callback a single argument which is either C<undef> or an array-ref
341with the filenames.
342
343=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
344
345Scans a directory (similar to C<aio_readdir>) but additionally tries to
346separate the entries of directory C<$path> into two sets of names, ones
347you can recurse into (directories or links to them), and ones you cannot
348recurse into (everything else).
349
350C<aio_scandir> is a composite request that consists of many sub
351requests. C<$maxreq> specifies the maximum number of outstanding aio
352requests that this function generates. If it is C<< <= 0 >>, then a
353suitable default will be chosen (currently 8).
354
355On error, the callback is called without arguments, otherwise it receives
356two array-refs with path-relative entry names.
357
358Example:
359
360 aio_scandir $dir, 0, sub {
361 my ($dirs, $nondirs) = @_;
362 print "real directories: @$dirs\n";
363 print "everything else: @$nondirs\n";
364 };
365
366Implementation notes.
367
368The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.
369
370After reading the directory, the modification time, size etc. of the
371directory before and after the readdir is checked, and if they match (and
372isn't the current time), the link count will be used to decide how many
373entries are directories (if >= 2). Otherwise, no knowledge of the number
374of subdirectories will be assumed.
375
376Then entries will be sorted into likely directories (everything without
377a non-initial dot currently) and likely non-directories (everything
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
382filesystems might detect the type of the entry without reading the inode
383data (e.g. ext2fs filetype feature).
384
385If the known number of directories (link count - 2) has been reached, the
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.
394
395=cut
396
397sub aio_scandir($$$) {
398 my ($path, $maxreq, $cb) = @_;
399
400 my $grp = aio_group;
401
402 $maxreq = 8 if $maxreq <= 0;
403
404 # stat once
405 add $grp aio_stat $path, sub {
406 return $cb->() if $_[0];
407 my $now = time;
408 my $hash1 = join ":", (stat _)[0,1,3,7,9];
409
410 # read the directory entries
411 add $grp aio_readdir $path, sub {
412 my $entries = shift
413 or return $cb->();
414
415 # stat the dir another time
416 add $grp aio_stat $path, sub {
417 my $hash2 = join ":", (stat _)[0,1,3,7,9];
418
419 my $ndirs;
420
421 # take the slow route if anything looks fishy
422 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
423 $ndirs = -1;
424 } else {
425 # if nlink == 2, we are finished
426 # on non-posix-fs's, we rely on nlink < 2
427 $ndirs = (stat _)[3] - 2
428 or return $cb->([], $entries);
429 }
430
431 # sort into likely dirs and likely nondirs
432 # dirs == files without ".", short entries first
433 $entries = [map $_->[0],
434 sort { $b->[1] cmp $a->[1] }
435 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
436 @$entries];
437
438 my (@dirs, @nondirs);
439
440 my ($statcb, $schedcb);
441 my $nreq = 0;
442
443 $schedcb = sub {
444 if (@$entries) {
445 if ($nreq < $maxreq) {
446 my $ent = pop @$entries;
447 $nreq++;
448 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
449 }
450 } elsif (!$nreq) {
451 # finished
452 undef $statcb;
453 undef $schedcb;
454 $cb->(\@dirs, \@nondirs) if $cb;
455 undef $cb;
456 }
457 };
458 $statcb = sub {
459 my ($status, $entry) = @_;
460
461 if ($status < 0) {
462 $nreq--;
463 push @nondirs, $entry;
464 &$schedcb;
465 } else {
466 # need to check for real directory
467 add $grp aio_lstat "$path/$entry", sub {
468 $nreq--;
469
470 if (-d _) {
471 push @dirs, $entry;
472
473 if (!--$ndirs) {
474 push @nondirs, @$entries;
475 $entries = [];
476 }
477 } else {
478 push @nondirs, $entry;
479 }
480
481 &$schedcb;
482 }
483 }
484 };
485
486 &$schedcb while @$entries && $nreq < $maxreq;
487 };
488 };
489 };
490
491 $grp
492}
493
236=item aio_fsync $fh, $callback 494=item aio_fsync $fh, $callback->($status)
237 495
238Asynchronously call fsync on the given filehandle and call the callback 496Asynchronously call fsync on the given filehandle and call the callback
239with the fsync result code. 497with the fsync result code.
240 498
241=item aio_fdatasync $fh, $callback 499=item aio_fdatasync $fh, $callback->($status)
242 500
243Asynchronously call fdatasync on the given filehandle and call the 501Asynchronously call fdatasync on the given filehandle and call the
244callback with the fdatasync result code. 502callback with the fdatasync result code.
245 503
246If 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
247detected, 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.
248 612
249=back 613=back
250 614
251=head2 SUPPORT FUNCTIONS 615=head2 SUPPORT FUNCTIONS
252 616
380} 744}
381 745
3821; 7461;
383 747
384=head2 FORK BEHAVIOUR 748=head2 FORK BEHAVIOUR
749
750This module should do "the right thing" when the process using it forks:
385 751
386Before the fork, IO::AIO enters a quiescent state where no requests 752Before the fork, IO::AIO enters a quiescent state where no requests
387can 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
388the fork the parent simply leaves the quiescent state and continues 754the fork the parent simply leaves the quiescent state and continues
389request/result processing, while the child clears the request/result 755request/result processing, while the child clears the request/result
390queue (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
391the 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
392parent process has been reached again. 758parent process has been reached again.
393 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
394=head1 SEE ALSO 764=head1 SEE ALSO
395 765
396L<Coro>, L<Linux::AIO>. 766L<Coro>, L<Linux::AIO> (obsolete).
397 767
398=head1 AUTHOR 768=head1 AUTHOR
399 769
400 Marc Lehmann <schmorp@schmorp.de> 770 Marc Lehmann <schmorp@schmorp.de>
401 http://home.schmorp.de/ 771 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines